[Flink] fix lazy source enumerator behaviour on error

jto · jto · commit e359edacfe2e · 2024-11-13T16:57:22.000+01:00
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/LazyFlinkSourceSplitEnumerator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/LazyFlinkSourceSplitEnumerator.java
@@ -84,6 +84,7 @@ public void start() {
           }
         },
         (sourceSplits, error) -> {
+          pendingSplits.addAll(sourceSplits);
           if (error != null) {
             throw new RuntimeException("Failed to start source enumerator.", error);
           }

Original file line number	Diff line number	Diff line change
`@@ -84,6 +84,7 @@ public void start() {`
`84`	`84`	`}`
`85`	`85`	`},`
`86`	`86`	`(sourceSplits, error) -> {`
	`87`	`+ pendingSplits.addAll(sourceSplits);`
`87`	`88`	`if (error != null) {`
`88`	`89`	`throw new RuntimeException("Failed to start source enumerator.", error);`
`89`	`90`	`}`