apache · bowenli86 · Mar 30, 2026
diff --git a/...reaming/api/operators/python/embedded/AbstractOneInputEmbeddedPythonFunctionOperator.java b/...reaming/api/operators/python/embedded/AbstractOneInputEmbeddedPythonFunctionOperator.java
@@ -31,7 +31,6 @@
 import org.apache.flink.util.Preconditions;
 
 import com.google.protobuf.AbstractMessageLite;
-import pemja.core.object.PyIterator;
 
 import java.util.List;
 import java.util.stream.Collectors;
@@ -151,18 +150,17 @@ public void processElement(StreamRecord<IN> element) throws Exception {
         timestamp = element.getTimestamp();
 
         IN value = element.getValue();
-        PyIterator results =
-                (PyIterator)
+        try (EmbeddedPythonIterator results =
+                EmbeddedPythonIterator.from(
                         interpreter.invokeMethod(
                                 "operation",
                                 "process_element",
-                                inputDataConverter.toExternal(value));
-
-        while (results.hasNext()) {
-            OUT result = outputDataConverter.toInternal(results.next());
-            collector.collect(result);
+                                inputDataConverter.toExternal(value)))) {
+            while (results.hasNext()) {
+                OUT result = outputDataConverter.toInternal(results.next());
+                collector.collect(result);
+            }
         }
-        results.close();
     }
 
     TypeInformation<IN> getInputTypeInfo() {

diff --git a/...reaming/api/operators/python/embedded/AbstractTwoInputEmbeddedPythonFunctionOperator.java b/...reaming/api/operators/python/embedded/AbstractTwoInputEmbeddedPythonFunctionOperator.java
@@ -31,7 +31,6 @@
 import org.apache.flink.util.Preconditions;
 
 import com.google.protobuf.AbstractMessageLite;
-import pemja.core.object.PyIterator;
 
 import java.util.List;
 import java.util.stream.Collectors;
@@ -178,18 +177,17 @@ public void processElement1(StreamRecord<IN1> element) throws Exception {
         timestamp = element.getTimestamp();
 
         IN1 value = element.getValue();
-        PyIterator results =
-                (PyIterator)
+        try (EmbeddedPythonIterator results =
+                EmbeddedPythonIterator.from(
                         interpreter.invokeMethod(
                                 "operation",
                                 "process_element1",
-                                inputDataConverter1.toExternal(value));
-
-        while (results.hasNext()) {
-            OUT result = outputDataConverter.toInternal(results.next());
-            collector.collect(result);
+                                inputDataConverter1.toExternal(value)))) {
+            while (results.hasNext()) {
+                OUT result = outputDataConverter.toInternal(results.next());
+                collector.collect(result);
+            }
         }
-        results.close();
     }
 
     @Override
@@ -198,18 +196,17 @@ public void processElement2(StreamRecord<IN2> element) throws Exception {
         timestamp = element.getTimestamp();
 
         IN2 value = element.getValue();
-        PyIterator results =
-                (PyIterator)
+        try (EmbeddedPythonIterator results =
+                EmbeddedPythonIterator.from(
                         interpreter.invokeMethod(
                                 "operation",
                                 "process_element2",
-                                inputDataConverter2.toExternal(value));
-
-        while (results.hasNext()) {
-            OUT result = outputDataConverter.toInternal(results.next());
-            collector.collect(result);
+                                inputDataConverter2.toExternal(value)))) {
+            while (results.hasNext()) {
+                OUT result = outputDataConverter.toInternal(results.next());
+                collector.collect(result);
+            }
         }
-        results.close();
     }
 
     TypeInformation<IN1> getInputTypeInfo1() {

diff --git a/...java/org/apache/flink/streaming/api/operators/python/embedded/EmbeddedPythonIterator.java b/...java/org/apache/flink/streaming/api/operators/python/embedded/EmbeddedPythonIterator.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.python.embedded;
+
+import org.apache.flink.annotation.Internal;
+
+import java.lang.reflect.Method;
+import java.util.Objects;
+
+/**
+ * Reflective adapter for embedded Python iterators.
+ *
+ * <p>PEMJA iterator objects may come back from a different user-code classloader after recovery, so
+ * callers should not hard-cast them to {@code pemja.core.object.PyIterator}.
+ */
+@Internal
+public final class EmbeddedPythonIterator implements AutoCloseable {
+
+    private final Object iterator;
+    private final Method hasNextMethod;
+    private final Method nextMethod;
+    private final Method closeMethod;
+
+    private EmbeddedPythonIterator(Object iterator) {
+        this.iterator = Objects.requireNonNull(iterator, "iterator must not be null");
+
+        try {
+            Class<?> iteratorClass = iterator.getClass();
+            this.hasNextMethod = iteratorClass.getMethod("hasNext");
+            this.nextMethod = iteratorClass.getMethod("next");
+            this.closeMethod = iteratorClass.getMethod("close");
+        } catch (ReflectiveOperationException e) {
+            throw new IllegalStateException(
+                    String.format(
+                            "Failed to adapt embedded Python iterator of type %s.",
+                            iterator.getClass().getName()),
+                    e);
+        }
+    }
+
+    public static EmbeddedPythonIterator from(Object iterator) {
+        return new EmbeddedPythonIterator(iterator);
+    }
+
+    public boolean hasNext() throws Exception {
+        return (boolean) hasNextMethod.invoke(iterator);
+    }
+
+    public Object next() throws Exception {
+        return nextMethod.invoke(iterator);
+    }
+
+    @Override
+    public void close() throws Exception {
+        closeMethod.invoke(iterator);
+    }
+}
diff --git a/...e/flink/streaming/api/operators/python/embedded/EmbeddedPythonKeyedCoProcessOperator.java b/...e/flink/streaming/api/operators/python/embedded/EmbeddedPythonKeyedCoProcessOperator.java
@@ -36,8 +36,6 @@
 import org.apache.flink.streaming.api.utils.PythonTypeUtils;
 import org.apache.flink.types.Row;
 
-import pemja.core.object.PyIterator;
-
 import java.util.List;
 
 import static org.apache.flink.python.PythonOptions.MAP_STATE_READ_CACHE_SIZE;
@@ -149,15 +147,14 @@ private void invokeUserFunction(TimeDomain timeDomain, InternalTimer<K, VoidName
         onTimerContext.timeDomain = timeDomain;
         onTimerContext.timer = timer;
 
-        PyIterator results =
-                (PyIterator)
-                        interpreter.invokeMethod("operation", "on_timer", timer.getTimestamp());
-
-        while (results.hasNext()) {
-            OUT result = outputDataConverter.toInternal(results.next());
-            collector.collect(result);
+        try (EmbeddedPythonIterator results =
+                EmbeddedPythonIterator.from(
+                        interpreter.invokeMethod("operation", "on_timer", timer.getTimestamp()))) {
+            while (results.hasNext()) {
+                OUT result = outputDataConverter.toInternal(results.next());
+                collector.collect(result);
+            }
         }
-        results.close();
 
         onTimerContext.timeDomain = null;
         onTimerContext.timer = null;

diff --git a/...che/flink/streaming/api/operators/python/embedded/EmbeddedPythonKeyedProcessOperator.java b/...che/flink/streaming/api/operators/python/embedded/EmbeddedPythonKeyedProcessOperator.java
@@ -36,8 +36,6 @@
 import org.apache.flink.streaming.api.utils.PythonTypeUtils;
 import org.apache.flink.types.Row;
 
-import pemja.core.object.PyIterator;
-
 import java.util.List;
 
 import static org.apache.flink.python.PythonOptions.MAP_STATE_READ_CACHE_SIZE;
@@ -143,15 +141,14 @@ private void invokeUserFunction(TimeDomain timeDomain, InternalTimer<K, VoidName
             throws Exception {
         onTimerContext.timeDomain = timeDomain;
         onTimerContext.timer = timer;
-        PyIterator results =
-                (PyIterator)
-                        interpreter.invokeMethod("operation", "on_timer", timer.getTimestamp());
-
-        while (results.hasNext()) {
-            OUT result = outputDataConverter.toInternal(results.next());
-            collector.collect(result);
+        try (EmbeddedPythonIterator results =
+                EmbeddedPythonIterator.from(
+                        interpreter.invokeMethod("operation", "on_timer", timer.getTimestamp()))) {
+            while (results.hasNext()) {
+                OUT result = outputDataConverter.toInternal(results.next());
+                collector.collect(result);
+            }
         }
-        results.close();
 
         onTimerContext.timeDomain = null;
         onTimerContext.timer = null;

diff --git a/...rg/apache/flink/streaming/api/operators/python/embedded/EmbeddedPythonWindowOperator.java b/...rg/apache/flink/streaming/api/operators/python/embedded/EmbeddedPythonWindowOperator.java
@@ -34,8 +34,6 @@
 import org.apache.flink.table.runtime.operators.window.Window;
 import org.apache.flink.types.Row;
 
-import pemja.core.object.PyIterator;
-
 import java.util.List;
 
 import static org.apache.flink.python.PythonOptions.MAP_STATE_READ_CACHE_SIZE;
@@ -143,15 +141,14 @@ public <T> DataStreamPythonFunctionOperator<T> copy(
     private void invokeUserFunction(InternalTimer<K, W> timer) throws Exception {
         windowTimerContext.timer = timer;
 
-        PyIterator results =
-                (PyIterator)
-                        interpreter.invokeMethod("operation", "on_timer", timer.getTimestamp());
-
-        while (results.hasNext()) {
-            OUT result = outputDataConverter.toInternal(results.next());
-            collector.collect(result);
+        try (EmbeddedPythonIterator results =
+                EmbeddedPythonIterator.from(
+                        interpreter.invokeMethod("operation", "on_timer", timer.getTimestamp()))) {
+            while (results.hasNext()) {
+                OUT result = outputDataConverter.toInternal(results.next());
+                collector.collect(result);
+            }
         }
-        results.close();
 
         windowTimerContext.timer = null;
     }

diff --git a/...pache/flink/table/runtime/operators/python/table/EmbeddedPythonTableFunctionOperator.java b/...pache/flink/table/runtime/operators/python/table/EmbeddedPythonTableFunctionOperator.java
@@ -22,6 +22,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.fnexecution.v1.FlinkFnApi;
 import org.apache.flink.python.util.ProtoUtils;
+import org.apache.flink.streaming.api.operators.python.embedded.EmbeddedPythonIterator;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
@@ -33,8 +34,6 @@
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.util.Preconditions;
 
-import pemja.core.object.PyIterator;
-
 import static org.apache.flink.python.PythonOptions.PYTHON_METRIC_ENABLED;
 import static org.apache.flink.python.PythonOptions.PYTHON_PROFILE_ENABLED;
 import static org.apache.flink.python.util.ProtoUtils.createFlattenRowTypeCoderInfoDescriptorProto;
@@ -147,25 +146,25 @@ public void processElement(StreamRecord<RowData> element) throws Exception {
                     userDefinedFunctionInputConverters[i].toExternal(value, udfInputOffsets[i]);
         }
 
-        PyIterator udtfResults =
-                (PyIterator)
+        try (EmbeddedPythonIterator udtfResults =
+                EmbeddedPythonIterator.from(
                         interpreter.invokeMethod(
                                 "table_operation",
                                 "process_element",
-                                (Object) (userDefinedFunctionInputArgs));
-
-        if (udtfResults.hasNext()) {
-            do {
-                Object[] udtfResult = (Object[]) udtfResults.next();
-                for (int i = 0; i < udtfResult.length; i++) {
-                    reuseResultRowData.setField(
-                            i, userDefinedFunctionOutputConverters[i].toInternal(udtfResult[i]));
-                }
-                rowDataWrapper.collect(reuseJoinedRow.replace(value, reuseResultRowData));
-            } while (udtfResults.hasNext());
-        } else if (joinType == FlinkJoinType.LEFT) {
-            rowDataWrapper.collect(reuseJoinedRow.replace(value, reuseNullResultRowData));
+                                (Object) (userDefinedFunctionInputArgs)))) {
+            if (udtfResults.hasNext()) {
+                do {
+                    Object[] udtfResult = (Object[]) udtfResults.next();
+                    for (int i = 0; i < udtfResult.length; i++) {
+                        reuseResultRowData.setField(
+                                i,
+                                userDefinedFunctionOutputConverters[i].toInternal(udtfResult[i]));
+                    }
+                    rowDataWrapper.collect(reuseJoinedRow.replace(value, reuseResultRowData));
+                } while (udtfResults.hasNext());
+            } else if (joinType == FlinkJoinType.LEFT) {
+                rowDataWrapper.collect(reuseJoinedRow.replace(value, reuseNullResultRowData));
+            }
         }
-        udtfResults.close();
     }
 }