Support SpeculativeExecution with Flink in batch execution mode

apache · Jun 29, 2024 · 347e25a · 347e25a
1 parent a47937c
commit 347e25a
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/...k/src/main/java/org/apache/iceberg/flink/source/enumerator/AbstractIcebergEnumerator.java b/...k/src/main/java/org/apache/iceberg/flink/source/enumerator/AbstractIcebergEnumerator.java
@@ -95,6 +95,13 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
     }
   }
 
+  // Flink's SourceCoordinator already keeps track of subTask to splits mapping.
+  // It already takes care of re-assigning splits to speculated attempts as well.
+  @Override
+  public void handleSourceEvent(int subTaskId, int attemptNumber, SourceEvent sourceEvent) {
+    handleSourceEvent(subTaskId, sourceEvent);
+  }
+
   @Override
   public void addSplitsBack(List<IcebergSourceSplit> splits, int subtaskId) {
     LOG.info("Add {} splits back to the pool for failed subtask {}", splits.size(), subtaskId);