[runtime][python] Release bridge-owned Pemja objects during cleanup

[joeyutong]

Linked issue: #942

Purpose of change

The embedded Python bridge owns several Pemja PyObject handles: the async thread pool and runner context in PythonActionExecutor, the Python resource context in PythonResourceAdapterImpl, and the Python Mem0 object in Mem0LongTermMemory. Each handle keeps a native Python reference, while Python objects can retain the surrounding Java task graph through Pemja proxies and JNI global references.

The existing close paths either performed only Python-level cleanup or did not close the PyObject at all. Repeated task failovers could therefore retain handles and old task graphs after the corresponding attempt had closed.

This change:

• runs Python-level cleanup before releasing each owned native handle;
• clears handle fields before releasing them, making repeated close calls safe against Pemja's unguarded native decRef;
• closes Mem0, the action executor, the Python resource adapter, the interpreter, and the environment in reverse creation order;
• continues closing sibling and upper-level resources after a cleanup failure while preserving the first exception and suppressing later ones; and
• clears the Python runner context's long-term-memory reference before cleanup, breaking the FlinkRunnerContext -> Mem0LongTermMemory -> FlinkRunnerContext cycle so native-handle release can collapse the full Java -> Python -> Java retention chain.

The PythonBridgeManager fallback also covers subtasks that initialized Mem0 but never created a Java RunnerContext, so cleanup does not depend on whether that subtask processed a Java action.

Tests

• Java lifecycle and failure-path tests: 55 passed across PythonActionExecutorTest, PythonResourceAdapterImplTest, Mem0LongTermMemoryTest, PythonBridgeManagerTest, and ActionExecutionOperatorTest.
• Python runner-context cleanup tests: 10 passed, including idempotent close and LTM cleanup failure paths.
• Ruff and Maven Spotless checks passed.
• Local 20-restart failover A/B for the original runner-context leak reduced post-Full-GC heap growth from 29.2 MiB to 1.7 MiB and removed the old runner-context/task graphs after termination.
• Focused HPROF A/B for the two sibling handles found 84 retained target PyObject handles after termination in each baseline; the native-close variants retained 0.

API

No user-facing API changes.

Documentation

• doc-needed
• doc-not-needed
• doc-included

[weiqingy]

Thanks for chasing this one down. I reverted close() locally and both new tests fail against the old body, so the regression coverage is real. A few questions inline.

[weiqingy]

nit: The copy-then-null reads as defensive style, but it looks load-bearing. PyObject.close() in pemja 0.5.7 is an unguarded decRef(tState, pyobject) with no null check and no double-close flag, so clearing the fields first is the only thing stopping a repeated close() from decrementing a second time on an already-released handle. Someone later tidying this into closePythonObject(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool) would drop that quietly, and the only assertion that would notice is the three-line tail of releasesBothPythonObjectsWhenLogicalCleanupFails.

Would a short comment here save the next reader that trip? There is precedent right next door at ActionExecutionOperator.java:471 (// Must close before pythonInterpreter since cached resources may hold Python references.).

Something like this, if it helps:

// Clear the fields before releasing: PyObject.close() is an unguarded native decRef,
// so a repeated close() must not reach the same handle twice.

[joeyutong]

Good point. I added a short comment explaining why the fields are cleared before releasing the Pemja handles, so a repeated close() cannot reach the same handle twice.

[weiqingy]

Combining both failures and rethrowing is the right call. What I keep looking at is what happens to this exception one frame up:

// PythonBridgeManager.close(), lines 292-302
if (pythonActionExecutor != null) { pythonActionExecutor.close(); }
if (pythonInterpreter != null) { pythonInterpreter.close(); }
if (pythonEnvironmentManager != null) { pythonEnvironmentManager.close(); }

That is a plain sequence, so on exactly the failure path this PR is built for, pythonInterpreter.close() never runs, and that is the release that tears down the interpreter owning every handle still outstanding. ActionExecutionOperator.close() (lines 469-489) has the same shape across its five closes.

To be clear, this is pre-existing. The old close() threw on a failed interpreter.invoke too, so nothing has regressed here. But given the stated goal is "releases both handles even if one cleanup operation fails", how do you see that goal holding one frame up? Carrying the same firstOrSuppressed pattern into PythonBridgeManager.close() would make the guarantee end-to-end, though I may be missing a reason the interpreter is fine to leak on that path.

[joeyutong]

Great catch. The guarantee did not hold one frame up. I extended best-effort cleanup through both PythonBridgeManager and ActionExecutionOperator, preserving close order and suppressing later failures, with tests at both layers.

[weiqingy]

This is the shape the whole fix turns on: logical cleanup inside, native release on the way out. Two other Pemja handles in the same lifecycle still have the pre-PR shape.

Mem0LongTermMemory.close() (Mem0LongTermMemory.java:137-140) calls adapter.callMethod(pyMem0, "close", Map.of()) and never pyMem0.close(), which is the old PythonActionExecutor.close() exactly. It is live rather than dead code: RunnerContextImpl.java:339-345 calls ltm.close(), and the handle comes from PythonBridgeManager.java:237.

PythonResourceAdapterImpl.pythonResourceContext (PythonResourceAdapterImpl.java:86,99) is built as interpreter.invoke(GET_RESOURCE_CONTEXT, this), so it is a Java object handed into Python, the same JNI-global-ref pattern #942 describes. That class has no close() at all, and PythonBridgeManager.close() never touches the adapter.

I read the PR and #942 as deliberately scoped to the action executor, so I am not suggesting you widen this one. Is a follow-up issue the plan for the sibling handles, or is there something that already releases those two that I have missed?

[joeyutong]

Good catch. Both handles are live, and the A/B heap dumps confirmed the same retention shape, so I widened this PR instead of opening a follow-up:

• PythonResourceAdapterImpl now owns and closes pythonResourceContext; nulling the field first makes repeated close safe.
• Mem0LongTermMemory now runs Python-level cleanup and always closes pyMem0, including when logical cleanup fails; it is also idempotent.
• PythonBridgeManager now closes Mem0 and the resource adapter before the interpreter/environment, while continuing all closes after failures. This also covers Mem0-initialized subtasks that never created a Java RunnerContext.
• FlinkRunnerContext.close() clears __ltm before cleanup and still closes the resource cache if LTM cleanup fails. This breaks the Python context/Mem0 cycle so releasing the Pemja handles can collapse the full Java -> Python -> Java retention chain.
• Added lifecycle, failure-path, and repeated-close tests. In focused 20-restart HPROF A/B runs, each sibling baseline retained 84 target PyObject handles after termination; the native-close variants retained 0.