emecii commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3790069954
##########
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:
You're right, and I'd rather pin the claim than trim it — fixed in 0c4df632.
`.satisfies(t -> assertThat(t.getSuppressed()).isEmpty())` is now on every
single-failure path, not just the two contract-3 tests: this one,
`ActionTaskContextManagerTest.closeClosesContinuationExecutorWhenRunnerContextFails`,
both `Error` tests, and
`ActionExecutionOperatorTest.closeClosesEveryComponentWhenAnEarlierCloseFails`
— the last being the case where the assertion says something slightly
different, namely that a healthy `super.close()` contributes nothing.
One caveat I should state rather than let the table imply otherwise: these
are fidelity guards, not regression tests. I tried to construct a mutation of
the current ladders that produces a spurious suppressed entry on a
single-failure path and could not — `firstOrSuppressed` guards
self-suppression, and a duplicated rung rethrows the same instance, so the
guard absorbs it. They pin what contract 3 claims; they do not catch a defect
that exists today. I've noted that in the PR body alongside the same caveat for
`closeReportsFirstResourceFailureWithLaterOnesSuppressed`.
Thanks for reading the table against the tests this carefully — that's twice
now it was claiming more than it checked.
##########
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:
No objection — done in 0c4df632, and I think it's more than a nit.
Your reasoning is exactly right on both halves, and I checked the second
half rather than assuming it: reverting the change leaves all ten tests in
`ActionTaskContextManagerTest` green, so it is provably a no-op today and no
test can distinguish it. It's a fail-open guard, so I added a comment recording
why the redundant call is deliberate — otherwise it reads like something to
simplify away.
That was the one aggregation point of seven not going through
`firstOrSuppressed`, so they're now uniform.
Looking at that site for the overwrite case you describe, I noticed it had
no two-failure test at all — the aggregation there was entirely unpinned, at
either rung. Added `closeReportsFirstFailureWithLaterOneSuppressed`, which
fails if the second rung overwrites instead of suppressing. That doesn't cover
the rung you flagged, since the failure it would drop cannot exist yet, but it
does mean the site is no longer relying on the other ladders' tests for its
aggregation behavior.
--
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]
