[HUDI-5456] Flink streaming read skips uncommitted instants (#7540)

Actually introduced by HUDI-5234.
apache · Dec 23, 2022 · 2a472f4 · 2a472f4
1 parent bdfaa4e
commit 2a472f4
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/...nk-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java b/...nk-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
@@ -53,6 +53,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -66,8 +67,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import scala.Serializable;
-
 import static org.apache.hudi.common.table.timeline.HoodieTimeline.GREATER_THAN;
 import static org.apache.hudi.common.table.timeline.HoodieTimeline.GREATER_THAN_OR_EQUALS;
 import static org.apache.hudi.common.table.timeline.HoodieTimeline.LESSER_THAN_OR_EQUALS;
@@ -482,7 +481,7 @@ public List<HoodieInstant> filterInstantsWithRange(
     HoodieTimeline completedTimeline = commitTimeline.filterCompletedInstants();
     if (issuedInstant != null) {
       // returns early for streaming mode
-      return commitTimeline
+      return completedTimeline
           .getInstantsAsStream()
           .filter(s -> HoodieTimeline.compareTimestamps(s.getTimestamp(), GREATER_THAN, issuedInstant))
           .collect(Collectors.toList());

diff --git a/...atasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java b/...atasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java
@@ -83,6 +83,14 @@ void testFilterInstantsWithRange() {
     List<HoodieInstant> instantRange3 = iis.filterInstantsWithRange(timeline, null);
     assertEquals(3, instantRange3.size());
     assertIterableEquals(Arrays.asList(commit1, commit2, commit3), instantRange3);
+
+    // add an inflight instant which should be excluded
+    HoodieInstant commit4 = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMPACTION_ACTION, "4");
+    timeline.createNewInstant(commit4);
+    timeline = timeline.reload();
+    assertEquals(4, timeline.getInstants().size());
+    List<HoodieInstant> instantRange4 = iis.filterInstantsWithRange(timeline, null);
+    assertEquals(3, instantRange4.size());
   }
 
 }