attempt partial commit only in case commit policy is partial commit (#4157)

Author: Blazer-007

gobblin-core-base/src/main/java/org/apache/gobblin/source/extractor/JobCommitPolicy.java

@@ -19,10 +19,12 @@
 
 import java.util.Properties;
 
+import com.typesafe.config.Config;
 import lombok.Getter;
 
 import org.apache.gobblin.configuration.ConfigurationKeys;
 import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.util.ConfigUtils;
 
 
 /**
@@ -107,4 +109,16 @@ public static JobCommitPolicy getCommitPolicy(Properties jobProps) {
   public static JobCommitPolicy getCommitPolicy(State state) {
     return forName(state.getProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY));
   }
+
+  /**
+   * Get a {@link JobCommitPolicy} through its name specified in configuration property
+   * {@link ConfigurationKeys#JOB_COMMIT_POLICY_KEY}.
+   *
+   * @param config a {@link Config} instance carrying job configuration properties
+   * @return a {@link JobCommitPolicy} with the given name specified in {@link ConfigurationKeys#JOB_COMMIT_POLICY_KEY}
+   */
+  public static JobCommitPolicy getCommitPolicy(Config config) {
+    return forName(ConfigUtils.getString(config, ConfigurationKeys.JOB_COMMIT_POLICY_KEY,
+        ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY));
+  }
 }

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java

@@ -30,6 +30,7 @@
 import io.temporal.workflow.Workflow;
 
 import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.source.extractor.JobCommitPolicy;
 import org.apache.gobblin.temporal.cluster.WorkerConfig;
 import org.apache.gobblin.temporal.ddm.util.TemporalWorkFlowUtils;
 import org.apache.gobblin.temporal.ddm.work.CommitStats;
@@ -79,7 +80,10 @@ private CommitStats performWork(WUProcessingSpec workSpec, final Properties prop
       log.error("ProcessWorkUnits failure - attempting partial commit before re-throwing exception", e);
 
       try {
-        performCommitIfAnyWorkUnitsProcessed(workSpec, searchAttributes, workunitsProcessed, props);// Attempt partial commit before surfacing the failure
+        if (shouldAttemptPartialCommit()) {
+          performCommitIfAnyWorkUnitsProcessed(workSpec, searchAttributes, workunitsProcessed,
+              props);// Attempt partial commit before surfacing the failure
+        }
       } catch (Exception commitException) {
         // Combine current and commit exception messages for a more complete context
         String combinedMessage = String.format(
@@ -98,6 +102,11 @@ private CommitStats performWork(WUProcessingSpec workSpec, final Properties prop
     return performCommitIfAnyWorkUnitsProcessed(workSpec, searchAttributes, workunitsProcessed, props);
   }
 
+  private boolean shouldAttemptPartialCommit() {
+    return JobCommitPolicy.getCommitPolicy(WorkerConfig.of(this).orElse(ConfigFactory.load()))
+        .isAllowPartialCommit();
+  }
+
   private CommitStats performCommitIfAnyWorkUnitsProcessed(WUProcessingSpec workSpec,
       Map<String, Object> searchAttributes, Optional<Integer> workunitsProcessed, Properties props) {
     //  we are only inhibiting commit when workunitsProcessed is actually known to be zero