joeyutong opened a new issue, #1048: URL: https://github.com/apache/flink-agents/issues/1048
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description [`PythonActionExecutor`](https://github.com/apache/flink-agents/blob/8c17640a45b02d1a8975050a9164e9c1565dcf18/runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java#L129-L201) creates several action-scoped Pemja `PyObject` handles but never closes them: 1. `convert_json_to_python_event` returns a wrapper for the Python `Event`; 2. an async Python action returns a coroutine wrapper, which is stored in the interpreter globals; 3. each coroutine poll calls `interpreter.get(...)` and receives another wrapper; 4. the interpreter-global coroutine reference remains after the action completes. Pemja's [`PyObject`](https://github.com/alibaba/pemja/blob/release-0.5.7/src/main/java/pemja/core/object/PyObject.java#L20-L54) implements `AutoCloseable`, and `close()` is the operation that performs the native `decRef`. Converting a Python object to a Java `PyObject` [increments its Python reference count](https://github.com/alibaba/pemja/blob/release-0.5.7/src/main/c/pemja/core/pyutils.c#L2003-L2008). Because the wrapper has no finalizer or cleaner, dropping the Java local variable does not release that native reference. The result is a native-memory leak proportional to the number of Python Action invocations. It is visible in the TaskManager process RSS and CPython live-object count, but not as an equivalent JVM heap or Flink managed-memory increase. The interpreter global and its temporary Java wrapper have separate ownership. `interpreter.set(...)` leaves a reference in the Python globals, so the temporary coroutine wrapper can be closed immediately after `set`; the global should remain only while the coroutine is pending and be deleted once it finishes. A minimal A/B loop performed 120 conversions of a 1 MiB Event JSON in the same interpreter: - with `PyObject.close()`: RSS warmed from about 102 MiB to 127 MiB, then remained stable; - without `PyObject.close()`: RSS grew approximately linearly from about 101 MiB to 240 MiB. In a longer single-interpreter validation, the unpatched path gained about 5.3 million CPython allocated blocks between Action calls 12 and 156. With the action-scoped handles released, calls 12 through 552 gained only 1,578 blocks and stayed on a plateau. The TaskManager and interpreter were not restarted during that validation, so executor-close behavior was not involved. This is related to, but distinct from, #942 and #944. Those cover executor-lifetime handles released when a task attempt closes; this issue covers temporary handles created repeatedly while the same executor remains open. Expected behavior: completing Python Actions should not retain their Event object graphs, temporary coroutine wrappers, polling wrappers, or completed interpreter-global coroutine references. ### How to reproduce 1. Start one Pemja interpreter and repeatedly execute a Python Action without restarting the TaskManager or interpreter. 2. Use Events with a non-trivial payload so retained object graphs are visible; a 1 MiB string payload is sufficient. 3. Sample TaskManager RSS and `sys.getallocatedblocks()` after regular intervals and force Python/Java GC before samples. 4. Observe that both the RSS floor and allocated-block count continue to rise with the number of Action calls. 5. Repeat while explicitly closing every `PyObject` returned by `invoke`, `call`, and `get`, and delete completed coroutine globals. The allocated-block count reaches a stable plateau instead. ### Version and environment - Flink Agents: `release-0.3`; the same ownership pattern is present on current `main` - Pemja: 0.5.7 - Flink: 2.2 - Java: 11 - Python: 3.11 - Linux TaskManager with an embedded CPython interpreter ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
