Merge pull request #15724 from cdapio/CDAP-21070-wrap-output-format

[CDAP-21070] Delegate all remaining methods of output committer
cdapio · Oct 21, 2024 · 72c3b10 · 72c3b10
2 parents 1013fbe + dc986cb
commit 72c3b10
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 0 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
@@ -20,6 +20,7 @@
 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 +60,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 +132,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));