Delegate all remaining methods of output committer

cdapio · Oct 21, 2024 · aeff034 · aeff034
1 parent 1013fbe
commit aeff034
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 3 deletions.
diff --git a/...spark-core-base/src/main/java/io/cdap/cdap/etl/spark/io/StageTrackingOutputCommitter.java b/...spark-core-base/src/main/java/io/cdap/cdap/etl/spark/io/StageTrackingOutputCommitter.java
@@ -16,10 +16,8 @@
 
 package io.cdap.cdap.etl.spark.io;
 
-import io.cdap.cdap.etl.api.exception.ErrorDetailsProvider;
-import io.cdap.cdap.etl.api.exception.ErrorPhase;
-import io.cdap.cdap.etl.common.ErrorDetails;
 import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.JobStatus.State;
 import org.apache.hadoop.mapreduce.OutputCommitter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 
@@ -59,6 +57,37 @@ public void setupJob(JobContext jobContext) {
     }
   }
 
+  @Override
+  @Deprecated
+  public void cleanupJob(JobContext jobContext) {
+    try {
+      delegate.cleanupJob(jobContext);
+    } catch (Exception e) {
+      throw ErrorDetails.handleException(e, stageName, errorDetailsProvider,
+        ErrorPhase.COMMITTING);
+    }
+  }
+
+  @Override
+  public void commitJob(JobContext jobContext) {
+    try {
+      delegate.commitJob(jobContext);
+    } catch (Exception e) {
+      throw ErrorDetails.handleException(e, stageName, errorDetailsProvider,
+        ErrorPhase.COMMITTING);
+    }
+  }
+
+  @Override
+  public void abortJob(JobContext jobContext, State state) {
+    try {
+      delegate.abortJob(jobContext, state);
+    } catch (Exception e) {
+      throw ErrorDetails.handleException(e, stageName, errorDetailsProvider,
+        ErrorPhase.COMMITTING);
+    }
+  }
+
   @Override
   public void setupTask(TaskAttemptContext taskAttemptContext) {
     try {
@@ -100,10 +129,31 @@ public void abortTask(TaskAttemptContext taskAttemptContext) {
   }
 
   @Override
+  @Deprecated
   public boolean isRecoverySupported() {
     return delegate.isRecoverySupported();
   }
 
+  @Override
+  public boolean isCommitJobRepeatable(JobContext jobContext) {
+    try {
+      return delegate.isCommitJobRepeatable(jobContext);
+    } catch (Exception e) {
+      throw ErrorDetails.handleException(e, stageName, errorDetailsProvider,
+        ErrorPhase.COMMITTING);
+    }
+  }
+
+  @Override
+  public boolean isRecoverySupported(JobContext jobContext) {
+    try {
+      return delegate.isRecoverySupported(jobContext);
+    } catch (Exception e) {
+      throw ErrorDetails.handleException(e, stageName, errorDetailsProvider,
+        ErrorPhase.COMMITTING);
+    }
+  }
+
   @Override
   public void recoverTask(TaskAttemptContext taskContext) {
     try {

diff --git a/...ator-spark-core-base/src/main/java/io/cdap/cdap/etl/spark/io/TrackingOutputCommitter.java b/...ator-spark-core-base/src/main/java/io/cdap/cdap/etl/spark/io/TrackingOutputCommitter.java
@@ -77,10 +77,21 @@ public void abortTask(TaskAttemptContext taskContext) throws IOException {
   }
 
   @Override
+  @Deprecated
   public boolean isRecoverySupported() {
     return delegate.isRecoverySupported();
   }
 
+  @Override
+  public boolean isCommitJobRepeatable(JobContext jobContext) throws IOException {
+    return delegate.isCommitJobRepeatable(jobContext);
+  }
+
+  @Override
+  public boolean isRecoverySupported(JobContext jobContext) throws IOException {
+    return delegate.isRecoverySupported(jobContext);
+  }
+
   @Override
   public void recoverTask(TaskAttemptContext taskContext) throws IOException {
     delegate.recoverTask(new TrackingTaskAttemptContext(taskContext));