sangkyoonnam opened a new pull request, #29494: URL: https://github.com/apache/flink/pull/29494
## What is the purpose of the change Backport of #27849 (7373dc9e2a3) to release-2.2 for [FLINK-39226](https://issues.apache.org/jira/browse/FLINK-39226), following the discussion on [FLINK-38500](https://issues.apache.org/jira/browse/FLINK-38500). The cherry-pick applies without conflicts. In thread mode, resubmitting a PyFlink job to the same session cluster can fail with `ClassCastException: pemja.core.object.PyIterator cannot be cast to pemja.core.object.PyIterator` when the returned iterator's class and the cast target are defined by different user-code classloaders. ## Brief change log - Replace the direct `PyIterator` casts in the embedded operators with the reflective `EmbeddedPythonIterator` wrapper - Backport `EmbeddedPythonIteratorTest` and `ForeignClassLoaderIterator` ## Verifying this change `./mvnw -pl flink-python clean verify` passes on JDK 17, including `EmbeddedPythonIteratorTest`, and `spotless:check` and `checkstyle:check` pass. With the `flink-python` jar built from this branch (built with `-Pjava11-target`) in a 2.2.1 distribution, the FLINK-38500 reproducer (macOS arm64, uv-managed Python) passes all three submissions on JDK 11; with the stock 2.2.1 jar in the same setup, submissions 2 and 3 fail with the `ClassCastException`. With the default Java 17-targeted jar, submission 1 passes on JDK 17; submissions 2 and 3 still fail in pemja 0.5.7 with `NoSuchFieldException: loadedLibraryNames`. [alibaba/pemja#100](https://github.com/alibaba/pemja/pull/100) is merged upstream; I'll rerun this reproducer against a pemja release containing it before opening the separate dependency-bump PR. I haven't tested JDK 21 on this branch. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): yes, embedded Python iterator calls go through reflection - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, iterator handling after recovery or resubmission - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) AI was used for the backport verification and this description; the code is an unchanged cherry-pick of Bowen Li's commit. Generated-by: Claude Code 2.1.296 (Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
