wenjin272 opened a new pull request, #1092: URL: https://github.com/apache/flink-agents/pull/1092
Linked issue: Closes #1087 ### Purpose of change Concurrent cross-language calls no longer enter one embedded `PythonInterpreter` from multiple threads, avoiding the native JVM crashes observed in async ChatModel and Mem0 paths without globally serializing calls. #### Runtime flow 1. Each Python-enabled operator subtask creates an owner interpreter and a `PythonInterpreterManager`. 2. Every Java-to-Python action, resource, adapter, and conversion call goes through that manager. 3. The mailbox owner and managed Java async workers execute inline on their own thread-local interpreters. Other callers use a stable lane in a bounded callback-worker pool. 4. Multi-step operations retaining Python objects complete within one interpreter operation. 5. Operator shutdown stops the managed async workers first. Each managed or callback worker closes its interpreter on the owning thread, then the owner interpreter and Python environment are closed. #### Key decisions - Use a hybrid routing model instead of a global interpreter lock: trusted Java async workers keep direct thread-local execution, while Python-originated callbacks are moved away from CPython-created threads. - Give each unmanaged source thread a stable callback lane so related opaque Python handles stay on one interpreter. Lane count reuses `num-async-threads`; threads and interpreters remain lazy. - Reuse Python resource instances across thread-confined interpreters rather than initializing one resource graph per worker. - Keep the fix inside Flink Agents; no Pemja or Flink release change is required. ### Behavioral Semantics #### Interaction decisions | Python bridge present | Calling context | Result | |---|---|---| | No | Any | No interpreter manager or callback worker is created. | | Yes | Mailbox/owner thread | Execute inline with the owner interpreter. | | Yes | Managed Java async worker | Lazily create and reuse that worker's interpreter; execute inline. | | Yes | Python-created or other unmanaged thread | Submit synchronously to that caller's stable callback lane. | | Yes | Callback worker re-entering Java-to-Python | Reuse its bound interpreter inline; do not submit recursively. | #### Behavioral contracts 1. Bridge calls on different execution threads do not concurrently use one `PythonInterpreter`. 2. Independent managed-worker and callback-lane calls can overlap; the fix does not globally serialize Python bridge traffic. 3. Python-to-Java-to-Python callbacks use bounded interpreter-owning workers and preserve caller-to-lane affinity. 4. Python object conversion, invocation, result retention, and nested Java-to-Python callbacks stay on the selected interpreter. 5. Interpreters are closed on their owning worker threads, accepted work completes before teardown, and calls after manager closure are rejected. 6. Plans without Python actions, Python resources, or Mem0 retain the existing no-Python-runtime path. #### Failure behavior - A callback operation or interpreter initialization failure is propagated; this layer does not retry or fall back to sharing another interpreter. A partially initialized interpreter is closed before the failure escapes. - An interrupted caller finishes an accepted callback before its interrupt status is restored, preventing premature native-state teardown. - Closing the manager from a non-owner thread, configuring a non-positive callback-worker count, or invoking after close raises a managed exception. - Shutdown attempts all interpreter cleanup and reports later cleanup failures as suppressed exceptions. External chat, embedding, and vector-store failures keep their existing behavior. ### Tests | Behavioral contract | Tests | |---|---| | Per-thread interpreter isolation | `PythonInterpreterManagerTest.bindsDifferentInterpretersToDifferentThreads`; `ConcurrentChatModelCrossLanguageTest` | | Concurrent execution without a global lock | `PythonInterpreterManagerTest.doesNotSerializeCallsMadeByDifferentThreads`; both concurrent ChatModel E2Es | | Bounded, stable Python-originated callback routing | `routesUnmanagedThreadCallsAwayFromTheCallerThread`; `boundsInterpretersCreatedForTransientUnmanagedThreads`; `reentrantCallbackCallUsesSameInterpreter`; Python concurrent ChatModel E2E | | Single-interpreter conversion and retention | `PythonActionExecutorTest.keepsActionConversionInvocationAndAwaitableOnCallingThreadsInterpreter`; `JavaResourceAdapterTest.convertsPythonChatMessageWithCallingThreadsInterpreter` | | Owning-thread lifecycle and post-close rejection | `managedJavaAsyncWorkerCreatesUsesAndClosesInterpreterOnItself`; `callbackInterpreterIsClosedByItsOwningWorker`; `closesEveryInterpreterAndRejectsLaterCalls`; `AsyncExecutorThreadFactoryTest` | | No-Python plan remains lazy | `PythonBridgeManagerTest.openIsNoOpWhenPlanHasNeitherPythonActionsNorResources` | Coverage includes overlapping Java calls, Python-created-thread re-entry, reentrancy, bounded reuse, result conversion, and shutdown. The JDK 21 runtime suite passed 778 tests with no failures or skips; focused tests passed 29/29; Java and Python concurrent E2Es passed 1/1 each; `tools/build.sh`, Spotless, and Ruff passed. Not verified: the re-enabled Mem0 test was not rerun against live Elasticsearch or Milvus because this environment lacked `ACTION_API_KEY`, `ES_HOST`, and `MILVUS_URI`. No Pemja source changes were made. <details> <summary>Implementation invariants and supporting evidence</summary> - The owner thread is captured when the manager is constructed; only that thread may close the manager. - Managed-worker identity is scoped to `AsyncExecutorThreadFactory` execution and removed after its exit cleanup. - Callback lanes are single-thread executors. Executor termination is followed by an actual worker-thread join so interpreter cleanup finishes before Python runtime teardown. - The lifecycle read lock covers accepted operations; close first excludes new work, then drains callback executors and aggregates cleanup failures. - Each newly created interpreter imports the same bridge modules before use. </details> ### API No public API or configuration key is added. Existing callers require no source changes. `num-async-threads` now also bounds the per-subtask callback lanes; unused lanes create neither threads nor interpreters. Java-only and no-Python plans retain their existing execution behavior, apart from stricter async-worker shutdown waiting. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: Codex 0.144.5 (GPT-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]
