[runtime][python] Release per-action Pemja objects

[joeyutong]

Linked issue: #1048

Purpose of change

PythonActionExecutor repeatedly creates Java-side Pemja PyObject handles while a Python Action is running. Those handles own native Python references and were previously dropped without calling close().

This change:

• closes the temporary Python Event and coroutine wrappers after each Action invocation;
• keeps a pending coroutine alive through the independent interpreter-global reference established by interpreter.set(...);
• closes the temporary wrapper returned by each interpreter.get(...) poll;
• deletes the interpreter-global coroutine reference after the coroutine completes;
• returns only the awaitable completion state from Python instead of materializing ignored yielded/returned values across JNI.

The last point avoids a recursive ownership problem: an Action output may contain arbitrary nested Python objects, while Java never consumes the output. Actions communicate with the runtime through emitted Events, so the narrow boundary is a boolean completion state.

This is complementary to #944: that PR handles executor-lifetime objects during attempt cleanup, while this PR handles objects created repeatedly during the lifetime of an open executor.

Tests

• PythonActionExecutorTest: 11 passed, covering synchronous completion, async storage, Action failure, set failure, pending/completed polling, and polling failure.
• test_function.py: 27 passed, including yielded and returned values being reduced to completion state.
• ./tools/lint.sh -c: passed for Python lint and all Java Spotless modules.
• Local Flink A/B: 12 fresh single-slot jobs covering six ownership boundaries in fixed/leak variants, 100 records with a 1 MiB payload each. Every fixed variant ended with zero tracked live objects and zero retained payload. Each leak variant retained exactly 100 affected objects or globals; payload-bearing variants retained 100 MiB. Full details are in [Draft] [Bug] PythonActionExecutor leaks per-action Pemja PyObject references #1048.

API

No user-facing API changes. PythonActionExecutor and call_python_awaitable are internal runtime bridge helpers. The helper now returns bool rather than (bool, output) because the output was not consumed. PyObject.close() failures continue through the existing ActionTask.invoke(...) throws Exception path.

Documentation

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

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Codex (GPT-5)