weiqingy commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3745804607
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/PythonBridgeManagerTest.java:
##########
@@ -59,4 +69,98 @@ 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.
+ *
+ * <p>Also pins the close order documented on the class, which is
load-bearing rather than
+ * incidental: {@link PythonActionExecutor#close()} calls back into the
interpreter, so it has
+ * to run before the interpreter is 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");
Review Comment:
Both this test and
`ActionTaskContextManagerTest.closeClosesContinuationExecutorWhenRunnerContextFails`
map to contract 3, which reads "unchanged in type and identity, with nothing
suppressed".
The `nothing suppressed` half isn't asserted anywhere in the PR. Every
`getSuppressed()` assertion in it (`:128`, `ResourceCacheTest.java:308`,
`ActionExecutionOperatorTest.java:688`) is a contract-4 non-empty check, so
nothing pins the array as empty on a single-failure path. The identity half I
raised earlier is covered now: `:154` uses `isSameAs`, and both paths rethrow
through the one site at `PythonBridgeManager.java:325`.
The code is fine either way, it's the table claiming a bit more than the
tests check. Either remedy costs a line, so whichever you prefer: a
`.satisfies(t -> assertThat(t.getSuppressed()).isEmpty())` here, or dropping
`with nothing suppressed` from contract 3?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTaskContextManager.java:
##########
@@ -322,15 +323,31 @@ void removePythonAwaitableRef(ActionTask actionTask) {
@Override
public void close() throws Exception {
+ // Close the continuation executor even when the runner context fails
to close. The first
+ // failure is rethrown with the later one suppressed.
+ //
+ // The ladder catches Throwable, not Exception, so a non-Exception
Throwable from the
+ // runner context cannot strand the executor's thread pool. Neither
type implements
+ // AutoCloseable, so the aggregation is spelled out rather than
delegated.
+ Throwable firstFailure = null;
if (runnerContext != null) {
try {
runnerContext.close();
+ } catch (Throwable t) {
+ firstFailure = t;
Review Comment:
nit: this is the one aggregation point in the PR that doesn't go through
`ExceptionUtils.firstOrSuppressed`. It's equivalent today, since `firstFailure`
is still null from `:332`, but it fails open: a close added above it later
would have its failure overwritten here rather than suppressed. Any objection
to `firstFailure = ExceptionUtils.firstOrSuppressed(t, firstFailure);` for
consistency with the other six?
--
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]
