weiqingy commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3742706633
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -299,14 +300,29 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
+ // Close every component even when an earlier one fails, so a failing
action executor
+ // cannot leak the interpreter or the environment manager. The first
failure is
+ // rethrown with the later ones suppressed.
+ //
+ // The ladder catches Throwable, not Exception, and IOUtils.closeAll
is deliberately not
+ // used: both stop at the first non-Exception Throwable without
closing what follows, and
+ // what follows here is the native Python state.
Review Comment:
Worth knowing about #944, which is open and takes the other path here: it
rewrites these same three `close()` methods with `IOUtils.closeAll(...)`, and
it touches all seven files this PR does.
Your reasoning in this comment holds up as far as I can tell. `closeAll`
defaults `suppressedException` to `Exception.class` and rethrows anything that
is not one before closing the rest (flink-core 2.3.0), so
`closeReleasesInterpreterAndEnvironmentWhenActionExecutorThrowsError` would
fail against #944's version of this method. The component lists differ too,
since that call also passes `longTermMemory` and `pythonResourceAdapter`.
So the two disagree on the mechanism, not just on the same seven files, and
whichever lands second inherits the other's decision. Either way the `Error`
case looks like the thing to settle first. How would you like to sequence them?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -553,24 +553,38 @@ public void waitInFlightEventsFinished() throws Exception
{
@Override
public void close() throws Exception {
- // Must close before pythonInterpreter since cached resources may hold
Python references.
- if (resourceCache != null) {
- resourceCache.close();
- }
- if (contextManager != null) {
- contextManager.close();
- }
- if (pythonBridge != null) {
- pythonBridge.close();
- }
- if (eventRouter != null) {
- eventRouter.close();
+ // Close every component even when an earlier one fails, so a failing
close cannot leak
+ // the components behind it or skip super.close(). The first failure
is rethrown with
+ // the later ones suppressed. Order is preserved: the resource cache
must close before
+ // pythonInterpreter since cached resources may hold Python references.
+ //
+ // The ladder catches Throwable, not Exception, and IOUtils.closeAll
is deliberately not
+ // used: both stop at the first non-Exception Throwable without
closing what follows,
+ // which is the very leak this method has to avoid.
+ Throwable firstFailure = null;
+ for (AutoCloseable closeable :
+ new AutoCloseable[] {
+ resourceCache, contextManager, pythonBridge, eventRouter,
durableExecManager
Review Comment:
Something I noticed while reading the order comment above. `resourceCache`
closes first because cached resources may hold Python references, so they need
to go before the interpreter.
`ResourceCache.close()` itself still catches only `Exception`
(`ResourceCache.java:148` and `:160`), which is the shape this comment argues
against. If a cached `Resource.close()` throws something that is not an
`Exception`, it propagates straight out: the remaining cached resources are
skipped, `cache.clear()` never runs, and `resourceContext.close()` never runs.
This ladder then catches it and closes `pythonBridge` anyway, so the
interpreter goes down while those resources still point at it.
Not a regression, and an `Error` out of `Resource.close()` is unlikely in
practice. What made me look was the PR body citing `ResourceCache.close()` as
prior art for the aggregation shape. Is it intentionally left as-is, or would
widening those two catches be in scope here?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -525,6 +529,65 @@ private static void replaceOperatorLtm(
ltmField.set(operator, ltm);
}
+ private static void replaceOperatorField(
+ ActionExecutionOperator<?, ?> operator, String name, Object value)
throws Exception {
+ Field field = ActionExecutionOperator.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(operator, value);
+ }
+
+ /**
+ * A failing component must not strand the ones behind it. This matters
most for {@code
+ * resourceCache}, which closes first and aggregates its own failures, and
for {@code
+ * pythonBridge}, which releases the embedded Python interpreter.
+ */
+ @Test
+ void closeClosesEveryComponentWhenAnEarlierCloseFails() throws Exception {
+ KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness
=
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new
ActionExecutionOperatorFactory(TestAgent.getAgentPlan(false), true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class));
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>)
testHarness.getOperator();
+
+ ResourceCache resourceCache = mock(ResourceCache.class);
+ ActionTaskContextManager contextManager =
mock(ActionTaskContextManager.class);
+ PythonBridgeManager pythonBridge = mock(PythonBridgeManager.class);
+ EventRouter<Long, Object> eventRouter = mock(EventRouter.class);
+ DurableExecutionManager durableExecManager =
mock(DurableExecutionManager.class);
+ doThrow(new IllegalStateException("resource cache close failed"))
+ .when(resourceCache)
+ .close();
+
+ replaceOperatorField(operator, "resourceCache", resourceCache);
+ replaceOperatorField(operator, "contextManager", contextManager);
+ replaceOperatorField(operator, "pythonBridge", pythonBridge);
+ replaceOperatorField(operator, "eventRouter", eventRouter);
+ replaceOperatorField(operator, "durableExecManager",
durableExecManager);
+
+ try {
+ assertThatThrownBy(operator::close)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("resource cache close failed");
+
+ // The components behind the failing one are still released.
+ verify(contextManager).close();
+ verify(pythonBridge).close();
+ verify(eventRouter).close();
+ verify(durableExecManager).close();
Review Comment:
The test table maps this one to contracts 1 and 7. Contract 1 is well
covered by the four `verify`s, but I cannot find anything here that looks at
`super.close()`, and there is no case where it fails, so the aggregation half
of contract 7 is not exercised.
I have not run this, so treat what follows as reasoning rather than a
result. Two changes both look like they would leave this test green: deleting
the try/catch at `ActionExecutionOperator.java:579-583`, or restoring just the
old shape for the super call (`if (firstFailure == null) super.close();`).
Neither the throw nor the four `verify`s depend on it, and the `finally`
empties the ladder before teardown.
That second one is the interesting case, since it is a partial revert of
what this PR fixes. `AbstractStreamOperator.close()` is
`stateHandler.dispose()`, so skipping it strands the state backends.
The "it still runs" half looks cheap to pin, because a side effect of
`dispose()` is visible once the throwing `close()` returns. Making it actually
fail is harder, since `super.close()` binds statically and a subclass cannot
intercept it. Would an assertion for the first half be worth it, or would you
rather drop contract 7 from the table?
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/PythonBridgeManagerTest.java:
##########
@@ -59,4 +67,92 @@ void openIsNoOpWhenPlanHasNeitherPythonActionsNorResources()
throws Exception {
assertThat(bridge.getPythonRunnerContext()).isNull();
}
}
+
+ /**
+ * A failing action executor must not strand the interpreter or the
environment manager: both
+ * hold native Python state that leaks for the lifetime of the TaskManager
if never closed.
+ */
+ @Test
+ void closeReleasesInterpreterAndEnvironmentWhenActionExecutorFails()
throws Exception {
+ PythonBridgeManager bridge = new PythonBridgeManager();
+ PythonActionExecutor actionExecutor = mock(PythonActionExecutor.class);
+ PythonInterpreter interpreter = mock(PythonInterpreter.class);
+ PythonEnvironmentManager environmentManager =
mock(PythonEnvironmentManager.class);
+ doThrow(new IllegalStateException("action executor close failed"))
+ .when(actionExecutor)
+ .close();
+
+ setField(bridge, "pythonActionExecutor", actionExecutor);
+ setField(bridge, "pythonInterpreter", interpreter);
+ setField(bridge, "pythonEnvironmentManager", environmentManager);
+
+ assertThatThrownBy(bridge::close)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("action executor close failed");
+
+ verify(interpreter).close();
+ verify(environmentManager).close();
Review Comment:
Contract 1 mentions the existing order, and I am curious how much weight
that clause is meant to carry. `verify()` does not check order, and none of the
new tests use `InOrder`, so swapping the array at
`PythonBridgeManager.java:311-314` to `{pythonInterpreter,
pythonActionExecutor, pythonEnvironmentManager}` leaves all three tests in this
file green. The action executor still throws, and it is still the first failure
even though it is no longer the first close.
Order does look load-bearing rather than incidental: the class javadoc at
`:70-71` documents reverse-of-creation order, and
`PythonActionExecutor.close()` calls into the interpreter twice
(`PythonActionExecutor.java:205-219`), which throws once the interpreter is
already closed.
Something like this in one of the three, if useful:
```java
InOrder inOrder = inOrder(actionExecutor, interpreter, environmentManager);
inOrder.verify(actionExecutor).close();
inOrder.verify(interpreter).close();
inOrder.verify(environmentManager).close();
```
One test would be enough to pin it. Does that seem worth it?
--
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]
