You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "joeyutong (via GitHub)" <gi...@apache.org> on 2026/07/31 07:20:23 UTC
[PR] [runtime][python] Release Pemja objects when closing action executor [flink-agents]
joeyutong opened a new pull request, #944:
URL: https://github.com/apache/flink-agents/pull/944
Linked issue: #942
### Purpose of change
`PythonActionExecutor` owns the Pemja `PyObject` handles for its Python async thread pool and runner context. The existing close path invokes Python-level cleanup but does not call `PyObject.close()`, which performs Pemja's native `decRef`. Repeated task restarts can therefore retain old runner contexts and their Flink task graphs through JNI global references.
This change:
- runs Python-level cleanup before releasing each owned `PyObject`;
- releases both handles even if one cleanup operation fails;
- preserves the first failure and attaches later failures as suppressed exceptions; and
- clears the fields so repeated close calls are idempotent.
### Tests
- `mvn -B --no-transfer-progress -pl runtime -am -Dtest=PythonActionExecutorTest -Dsurefire.failIfNoSpecifiedTests=false test` (2 passed)
- Maven Spotless formatting/checks passed as part of the Java verification
- Local 20-restart failover A/B: post-Full-GC heap growth changed from 29.2 MiB to 1.7 MiB, and old runner-context/task graphs were no longer retained after termination
### API
No public API changes.
### Documentation
- [ ] `doc-needed`
- [x] `doc-not-needed`
- [ ] `doc-included`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3745476785
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
- }
- if (pythonEnvironmentManager != null) {
- pythonEnvironmentManager.close();
- }
+ IOUtils.closeAll(
Review Comment:
Checked whether the 2-arg overload offers a way to keep the utility. It does not, and the reason matters. Against `flink-core-2.3.0`, three closeables, first throwing `OutOfMemoryError`:
```
closeAll(a1, a2, a3) -> java.lang.OutOfMemoryError
closed: a1=true a2=false a3=false
closeAll(asList(b1,b2,b3), Throwable.class) -> java.lang.Exception: java.lang.OutOfMemoryError
instanceof Error = false
closed: b1=true b2=true b3=true
ExceptionUtils.rethrowException(new OutOfMemoryError("boom"))
-> java.lang.OutOfMemoryError, instanceof Error = true
```
`Throwable.class` does close everything, but `closeAll(Iterable, Class)` casts the collected throwable only when it is already an `Exception` and otherwise does `new Exception(t)`. `Task.preProcessException` unwraps only `WrappingRuntimeException` before testing `isJvmFatalError(t) || t instanceof OutOfMemoryError` and halting the JVM, so a wrapped OOM stops registering as fatal and comes back as an ordinary task failure. So `rethrowException` passing the `Error` through unwrapped is doing work no `closeAll` variant can.
On scope, there is a fourth `closeAll` site outside your list: `PythonActionExecutor.close()` (`PythonActionExecutor.java:200`). It is the pointed one for this PR, since both `PyObject` fields are nulled before the aggregate runs, so an `Error` out of the async-thread-pool release leaves the runner-context handle unclosed and unreachable. That is the exact leak this PR exists to close. Should that site be in scope for whichever PR carries the ladder?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3886379261
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -752,14 +752,16 @@ def config(self) -> ReadableConfiguration:
@override
def close(self) -> None:
- if self.long_term_memory is not None:
- self.long_term_memory.close()
-
- if self.__resource_cache is not None:
- try:
- self.__resource_cache.close()
- finally:
- self.__resource_cache = None
+ ltm = self.__ltm
+ self.__ltm = None
+ try:
+ if ltm is not None:
+ ltm.close()
+ finally:
+ resource_cache = self.__resource_cache
+ self.__resource_cache = None
+ if resource_cache is not None:
+ resource_cache.close()
Review Comment:
Thanks again for flagging this failure-ordering gap. After #987 merged, I rebased #944 onto its cleanup ladders and updated this path to reuse the same `_failure_of` / `_first_or_logged` helpers. Ordinary `Exception`s now keep the first failure and log later failures, while `SystemExit` and `KeyboardInterrupt` propagate instead of being demoted behind an earlier cleanup failure. The tests now cover both ordinary double failure and the interpreter-exit case. Updated in f0bf1e86.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3725959384
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -75,6 +81,40 @@ void resetReconcilableFixtures() {
TestAgent.resetMixedRecoveryFixture();
}
+ @Test
+ void closeAttemptsAllResourcesAndSuppressesLaterFailures() throws Exception {
+ AgentPlan plan = new AgentPlan(new HashMap<>(), new HashMap<>());
+ ActionStateStore actionStateStore = mock(ActionStateStore.class);
+ ActionExecutionOperator<Object, Object> operator =
+ new ActionExecutionOperator<>(plan, true, null, null, actionStateStore);
+ ResourceCache resourceCache = mock(ResourceCache.class);
+ ActionTaskContextManager contextManager = mock(ActionTaskContextManager.class);
+ PythonBridgeManager pythonBridge = mock(PythonBridgeManager.class);
+ RuntimeException resourceFailure = new RuntimeException("resource cache close failed");
+ RuntimeException contextFailure = new RuntimeException("context manager close failed");
+ RuntimeException bridgeFailure = new RuntimeException("python bridge close failed");
+ RuntimeException durableFailure = new RuntimeException("durable manager close failed");
+
+ doThrow(resourceFailure).when(resourceCache).close();
+ doThrow(contextFailure).when(contextManager).close();
+ doThrow(bridgeFailure).when(pythonBridge).close();
+ doThrow(durableFailure).when(actionStateStore).close();
+ setPrivateField(operator, "resourceCache", resourceCache);
+ setPrivateField(operator, "contextManager", contextManager);
+ setPrivateField(operator, "pythonBridge", pythonBridge);
+
+ assertThatThrownBy(operator::close)
+ .isSameAs(resourceFailure)
+ .hasSuppressedException(contextFailure)
+ .hasSuppressedException(bridgeFailure)
+ .hasSuppressedException(durableFailure);
+ InOrder closeOrder = inOrder(resourceCache, contextManager, pythonBridge, actionStateStore);
Review Comment:
This exercises six closeables and verifies four. `eventRouter` and `super::close` are never injected or asserted.
I tried deleting each from the `closeAll(...)` list, and all 407 runtime-module tests stay green either way. Dropping `super::close` would silently stop disposing Flink operator state, which is the costlier of the two.
`eventRouter` looks cheap to cover: it is `private final transient`, and the `setPrivateField` helper this test already uses works on non-static final instance fields, so a throwing mock could join the `InOrder`. Any appetite for that?
`super.close()` I don't have a cheap way to assert, so I'd call that half a known gap rather than something to chase here.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/CloseableUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.operator;
+
+import org.apache.flink.util.LambdaUtil;
+
+import java.util.Arrays;
+
+final class CloseableUtils {
+
+ private CloseableUtils() {}
+
+ static void closeAll(AutoCloseable... closeables) throws Exception {
Review Comment:
nit: `org.apache.flink.util.IOUtils.closeAll(AutoCloseable...) throws Exception` already exists with the same signature, and ships in both flink-core 1.20.0 and 2.3.0, so this isn't a cross-version gap. It has the semantics this helper relies on: null-guards, skips null elements, aggregates through `ExceptionUtils.firstOrSuppressed`, rethrows. The repo already uses it at `runtime/src/main/java/.../feedback/Checkpoints.java:22,63`.
Moving the helper out of `OperatorUtils` makes sense, since that class does have a per-version variant under `dist/flink-1.20/`. Is there a reason to prefer a local helper over the Flink one?
If it did become `IOUtils.closeAll`, one knock-on: `CloseableUtils` is package-private in `operator`, so `PythonActionExecutor.close()` (`PythonActionExecutor.java:200-214`) still hand-rolls the same `firstOrSuppressed` accumulation and can't reach it. The public utility would let that collapse too.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
Review Comment:
Both frames check out. This is about a third instance rather than those two.
`ActionTaskContextManager.close()` (`runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTaskContextManager.java:319-330`) still has the round-1 shape:
```java
if (runnerContext != null) {
try { runnerContext.close(); } finally { runnerContext = null; }
}
if (continuationActionExecutor != null) {
continuationActionExecutor.close();
}
```
It is argument 2 of the operator's own new `closeAll(...)` list, so it sits inside the chain you just hardened. The throw path is this PR's own failure case: `RunnerContextImpl.close()` (`context/RunnerContextImpl.java:340-344`) calls `this.ltm.close()`, which reaches `Mem0LongTermMemory.close()` and a Pemja `invoke`.
The skipped close is not a no-op everywhere. The JDK 21 variant (`runtime/src/main/java21/.../async/ContinuationActionExecutor.java:151-153`) is `asyncExecutor.shutdownNow()`, while the JDK 11 one (`:59`) is empty. So on JDK 21 a failing LTM close leaks those threads.
Narrow exposure: JDK 21 only, and only when a close is already failing. Worth giving this frame the same treatment while you are in here?
One wrinkle if you do. `ContinuationActionExecutor.close()` does not declare `throws`, so it would need `implements AutoCloseable` or a lambda wrapper to go into the varargs list.
I checked the rest of the chain for the same shape and it is clean, so this looks like the last one rather than the first of many.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on PR #944:
URL: https://github.com/apache/flink-agents/pull/944#issuecomment-5461897089
Thanks for the reminder and continued review, @wenjin272. I have rebased #944 onto current `main` after #987 and updated it accordingly. The Java cleanup ladders retain #987’s `Throwable` handling while adding `longTermMemory` and `pythonResourceAdapter`; the Python runner-context path now reuses #987’s first-failure helpers with focused failure-ordering tests. I also refreshed the PR description and test results. The updated commit is f0bf1e86.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3726939847
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/CloseableUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.operator;
+
+import org.apache.flink.util.LambdaUtil;
+
+import java.util.Arrays;
+
+final class CloseableUtils {
+
+ private CloseableUtils() {}
+
+ static void closeAll(AutoCloseable... closeables) throws Exception {
Review Comment:
You're right. I had previously worried that the available utilities did not quite match the required aggregation semantics, and I was probably over-cautious when extracting the local helper. has exactly the behavior needed here. I replaced both local-helper call sites and the hand-rolled path with it, removed , and also used it for the new aggregation in 56795e15.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3740950034
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -752,14 +752,16 @@ def config(self) -> ReadableConfiguration:
@override
def close(self) -> None:
- if self.long_term_memory is not None:
- self.long_term_memory.close()
-
- if self.__resource_cache is not None:
- try:
- self.__resource_cache.close()
- finally:
- self.__resource_cache = None
+ ltm = self.__ltm
+ self.__ltm = None
+ try:
+ if ltm is not None:
+ ltm.close()
+ finally:
+ resource_cache = self.__resource_cache
+ self.__resource_cache = None
+ if resource_cache is not None:
+ resource_cache.close()
Review Comment:
If both `ltm.close()` and `resource_cache.close()` fail, Python's `finally` semantics make the resource-cache exception primary and leave the earlier LTM failure only in `__context__`. This contradicts the PR's stated behavior of preserving the first exception and suppressing later ones, and it differs from the Java `IOUtils.closeAll` paths. Could we explicitly aggregate these failures and add a test where both closes throw?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3726939847
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/CloseableUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.operator;
+
+import org.apache.flink.util.LambdaUtil;
+
+import java.util.Arrays;
+
+final class CloseableUtils {
+
+ private CloseableUtils() {}
+
+ static void closeAll(AutoCloseable... closeables) throws Exception {
Review Comment:
You are right. I previously worried that the available utilities did not quite match the required aggregation semantics, and I was probably over-cautious when extracting the local helper. `IOUtils.closeAll` has exactly the behavior needed here. I replaced both local-helper call sites and the hand-rolled `PythonActionExecutor` path with it, removed `CloseableUtils`, and also used it for the new `ActionTaskContextManager` aggregation in 56795e15.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3759218796
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
- }
- if (pythonEnvironmentManager != null) {
- pythonEnvironmentManager.close();
- }
+ IOUtils.closeAll(
Review Comment:
Thanks for the detailed analysis. I agree that IOUtils.closeAll does not provide the required behavior for non-Exception Throwables. Since #987 is the smaller and more general fix, I prefer to let it land first and then rebase #944 on top of it, adding longTermMemory and pythonResourceAdapter to the resulting cleanup ladders. I also noticed that #944 introduces another IOUtils.closeAll inside PythonActionExecutor.close(); I will address that Error path as part of the rebase.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release Pemja objects when closing action executor [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3696783616
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
Review Comment:
Combining both failures and rethrowing is the right call. What I keep looking at is what happens to this exception one frame up:
```java
// 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.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
Review Comment:
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:
```java
// Clear the fields before releasing: PyObject.close() is an unguarded native decRef,
// so a repeated close() must not reach the same handle twice.
```
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
+ }
+ }
- if (pythonRunnerContext != null) {
- try {
- interpreter.invoke(CLOSE_FLINK_RUNNER_CONTEXT, pythonRunnerContext);
- } finally {
- pythonRunnerContext = null;
- }
+ private void closePythonObject(String closeFunction, PyObject pythonObject) throws Exception {
+ if (pythonObject != null) {
+ try (pythonObject) {
Review Comment:
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?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release Pemja objects when closing action executor [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3701953926
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
Review Comment:
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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "emecii (via GitHub)" <gi...@apache.org>.
emecii commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3742785700
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
- }
- if (pythonEnvironmentManager != null) {
- pythonEnvironmentManager.close();
- }
+ IOUtils.closeAll(
Review Comment:
Following up on @weiqingy's note in #987 that these two PRs disagree on mechanism rather than merely touching the same seven files. I'm the author of #987, so flagging my own interest up front.
I checked the `Error` case empirically against `flink-core-2.3.0` rather than reading the source, and the concern holds. Three closeables that record whether `close()` ran, the first one throwing:
```
A (first throws OutOfMemoryError): thrown=java.lang.OutOfMemoryError: boom
A closed flags -> a1=true a2=false a3=false
B (first throws IllegalStateException): thrown=java.lang.IllegalStateException: boom
B closed flags -> b1=true b2=true b3=true
```
The varargs overload delegates to `closeAll(Iterable)`, which delegates to `closeAll(Iterable, Class<T>)` with `suppressedException = Exception.class`; that method rethrows anything not assignable to it *before* closing the rest. So `closeAll` continues past an `Exception` but stops dead on any non-`Exception` `Throwable`.
At this call site that means an `Error` out of `longTermMemory.close()` or `pythonActionExecutor.close()` leaves `pythonInterpreter` and `pythonEnvironmentManager` unclosed — the native Python state this PR exists to release, retained for the lifetime of the TaskManager JVM. The same shape appears at the other two sites:
- `ActionExecutionOperator.close()`: an `Error` from `resourceCache` skips `contextManager`, `pythonBridge`, `eventRouter`, `durableExecManager`, and the trailing `super::close`, so `stateHandler.dispose()` is skipped too.
- `ActionTaskContextManager.close()`: an `Error` from the runner context strands the continuation executor's thread pool.
Worth noting that `ResourceCache.close()` (`ResourceCache.java:148` and `:160`) currently catches only `Exception`, so an `Error` out of a cached `Resource.close()` propagates unchanged and is a concrete way to reach the first of those.
Would you consider replacing `closeAll` with a `catch (Throwable)` ladder that aggregates via `ExceptionUtils.firstOrSuppressed` and rethrows via `ExceptionUtils.rethrowException`? It keeps the same first-failure-wins-with-later-ones-suppressed semantics, and `rethrowException` passes both `Error` and `Exception` through unwrapped, so callers still see the original type and instance. That is the shape `KafkaActionStateStore.close()` (#948) and `ResourceCache.close()` already use in this module.
On sequencing, since we overlap on seven files: #987 is the smaller change and only rewrites the three `close()` methods. If it lands first, this PR's rebase becomes additive — `longTermMemory` and `pythonResourceAdapter` slot into ladders that already exist — instead of a mechanism swap in one direction or the other. I'm equally happy to go the other way and rework #987 as a follow-up on top of this if you'd prefer not to reshuffle. Mainly I'd like the two not to land opposite decisions on the `Error` case. What works best for you?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3759217003
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -752,14 +752,16 @@ def config(self) -> ReadableConfiguration:
@override
def close(self) -> None:
- if self.long_term_memory is not None:
- self.long_term_memory.close()
-
- if self.__resource_cache is not None:
- try:
- self.__resource_cache.close()
- finally:
- self.__resource_cache = None
+ ltm = self.__ltm
+ self.__ltm = None
+ try:
+ if ltm is not None:
+ ltm.close()
+ finally:
+ resource_cache = self.__resource_cache
+ self.__resource_cache = None
+ if resource_cache is not None:
+ resource_cache.close()
Review Comment:
Good catch. I replaced the try/finally propagation with explicit aggregation: the long-term-memory failure remains the primary exception, while a later resource-cache failure is retained in its __context__. The new double-failure test asserts both identities and also verifies that repeated close() remains a no-op. Fixed in fc4c0fb3.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3703612061
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
+ }
+ }
- if (pythonRunnerContext != null) {
- try {
- interpreter.invoke(CLOSE_FLINK_RUNNER_CONTEXT, pythonRunnerContext);
- } finally {
- pythonRunnerContext = null;
- }
+ private void closePythonObject(String closeFunction, PyObject pythonObject) throws Exception {
+ if (pythonObject != null) {
+ try (pythonObject) {
Review Comment:
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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3745476785
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
- }
- if (pythonEnvironmentManager != null) {
- pythonEnvironmentManager.close();
- }
+ IOUtils.closeAll(
Review Comment:
Checked whether the 2-arg overload offers a way to keep the utility. It does not, and the reason matters. Against `flink-core-2.3.0`, three closeables, first throwing `OutOfMemoryError`:
```
closeAll(a1, a2, a3) -> java.lang.OutOfMemoryError
closed: a1=true a2=false a3=false
closeAll(asList(b1,b2,b3), Throwable.class) -> java.lang.Exception: java.lang.OutOfMemoryError
instanceof Error = false
closed: b1=true b2=true b3=true
ExceptionUtils.rethrowException(new OutOfMemoryError("boom"))
-> java.lang.OutOfMemoryError, instanceof Error = true
```
`Throwable.class` does close everything, but `closeAll(Iterable, Class)` casts the collected throwable only when it is already an `Exception` and otherwise does `new Exception(t)`. `Task.preProcessException` unwraps only `WrappingRuntimeException` before testing `isJvmFatalError(t) || t instanceof OutOfMemoryError` and halting the JVM, so a wrapped OOM stops registering as fatal and comes back as an ordinary task failure. So `rethrowException` passing the `Error` through unwrapped is doing work no `closeAll` variant can.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3745497391
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/CloseableUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.operator;
+
+import org.apache.flink.util.LambdaUtil;
+
+import java.util.Arrays;
+
+final class CloseableUtils {
+
+ private CloseableUtils() {}
+
+ static void closeAll(AutoCloseable... closeables) throws Exception {
Review Comment:
Thanks for addressing the comments. Nothing further from me.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #944:
URL: https://github.com/apache/flink-agents/pull/944
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release Pemja objects when closing action executor [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3701951729
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
Review Comment:
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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3745476785
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
- }
- if (pythonEnvironmentManager != null) {
- pythonEnvironmentManager.close();
- }
+ IOUtils.closeAll(
Review Comment:
Checked whether the 2-arg overload offers a way to keep the utility. It does not, and the reason matters. Against `flink-core-2.3.0`, three closeables, first throwing `OutOfMemoryError`:
```
closeAll(a1, a2, a3) -> java.lang.OutOfMemoryError
closed: a1=true a2=false a3=false
closeAll(asList(b1,b2,b3), Throwable.class) -> java.lang.Exception: java.lang.OutOfMemoryError
instanceof Error = false
closed: b1=true b2=true b3=true
ExceptionUtils.rethrowException(new OutOfMemoryError("boom"))
-> java.lang.OutOfMemoryError, instanceof Error = true
```
`Throwable.class` does close everything, but `closeAll(Iterable, Class)` casts the collected throwable only when it is already an `Exception` and otherwise does `new Exception(t)`. `Task.preProcessException` unwraps only `WrappingRuntimeException` before testing `isJvmFatalError(t) || t instanceof OutOfMemoryError` and halting the JVM, so a wrapped OOM stops registering as fatal and comes back as an ordinary task failure. So `rethrowException` passing the `Error` through unwrapped is doing work no `closeAll` variant can.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][python] Release bridge-owned Pemja objects during cleanup [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #944:
URL: https://github.com/apache/flink-agents/pull/944#issuecomment-5449547799
#987 is now merged, you can go ahead and rebase @joeyutong.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org