[hotfix][state] Make state iterator adaptor tolerate null iterable input

Zakelly · Dec 9, 2024 · 0bad951 · 0bad951
1 parent fa9d909
commit 0bad951
Showing 1 changed file with 10 additions and 3 deletions.
diff --git a/...untime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/CompleteStateIterator.java b/...untime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/CompleteStateIterator.java
@@ -22,6 +22,8 @@
 import org.apache.flink.api.common.state.v2.StateIterator;
 import org.apache.flink.core.state.StateFutureUtils;
 
+import javax.annotation.Nullable;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -35,9 +37,14 @@ public class CompleteStateIterator<T> implements StateIterator<T> {
     final Iterator<T> iterator;
     final boolean empty;
 
-    public CompleteStateIterator(Iterable<T> iterable) {
-        this.iterator = iterable.iterator();
-        this.empty = !iterator.hasNext();
+    public CompleteStateIterator(@Nullable Iterable<T> iterable) {
+        if (iterable == null) {
+            this.iterator = Collections.emptyIterator();
+            this.empty = true;
+        } else {
+            this.iterator = iterable.iterator();
+            this.empty = !iterator.hasNext();
+        }
     }
 
     @Override