[hotfix][runtime] Close every component when an earlier close fails

[emecii]

Linked issue: none (hotfix)

Purpose of change

Four close() methods in the operator shutdown chain closed their components as
sequential statements with no per-call guard, so the first failure skipped
everything behind it.

Site	Stranded when an earlier close fails
ActionExecutionOperator.close()	contextManager, pythonBridge, eventRouter, durableExecManager, super.close()
PythonBridgeManager.close()	the Pemja PythonInterpreter and the PythonEnvironmentManager
ActionTaskContextManager.close()	the ContinuationActionExecutor thread pool
ResourceCache.close()	the remaining cached resources, cache.clear(), and resourceContext.close() — for a non-Exception Throwable only

Runtime flow

ActionExecutionOperator.close() is the entry point and the sharpest case. It
closes resourceCache first, and ResourceCache.close() aggregates its own
component failures and rethrows them by design. So the exception ResourceCache
propagates is precisely the one that skipped the rest of the chain, including
pythonBridge.close() — the call that releases the embedded Python interpreter
and its environment manager. A single resource failing to close could leak native
Python state for the lifetime of the TaskManager JVM.

PythonBridgeManager.close() and ActionTaskContextManager.close() are reached
from that same chain and had the same shape.

ResourceCache.close() was added to this list during review. It already
aggregated Exceptions, but caught only Exception, so a non-Exception
Throwable out of a cached Resource.close() propagated immediately. That gap
matters more after the rest of this patch, not less: the operator ladder now
continues past the failure and closes the Python interpreter, while cached
resources that may hold Python references are still open — which inverts the
ordering the ladder exists to preserve. Thanks to @weiqingy for catching it.

Key decisions

Capture and rethrow, rather than closing later components in a finally. A
finally that completes abruptly discards the in-flight exception (JLS 14.20.2),
which is the defect #974 is fixing in FlussActionStateStore. The shape here
matches KafkaActionStateStore.close() (#948), and ResourceCache.close()
already aggregated this way before this patch widened it.

The ladders catch Throwable, not Exception. A catch (Exception) ladder
stops at a non-Exception Throwable and skips the remaining closes, which is
the same leak with a narrower trigger. ExceptionUtils.rethrowException then
rethrows Error and Exception unchanged, so the caller sees the original type
and instance rather than a wrapper.

IOUtils.closeAll was considered and rejected, for the reason already set out in
#974: with the default Exception.class it rethrows a non-Exception
Throwable immediately without closing the remaining resources. I verified this
against flink-core-2.3.0 rather than assuming — with an Error thrown from the
first closeable, the second is never closed, while a plain Exception closes all
three. An earlier revision of this patch used closeAll and the Error test
below is what caught it. The same finding is recorded on #944, which takes the
closeAll path over these same files.

ActionTaskContextManager spells the aggregation out rather than delegating:
neither RunnerContextImpl nor ContinuationActionExecutor implements
AutoCloseable, and ContinuationActionExecutor has separate java/ and
java21/ implementations, so making it closeable would touch a source set the
JDK 11 profile does not compile.

Behavioral contracts

1. Every component close is attempted on every call, in the existing order.
2. A null component is skipped rather than raising.
3. When one close fails, its exception reaches the caller unchanged in type and
   identity, with nothing suppressed.
4. When several fail, the first is thrown and the later ones are attached to it
   via addSuppressed.
5. A non-Exception Throwable does not prevent the remaining closes, and
   reaches the caller as itself rather than wrapped.
6. When nothing fails, close() returns normally.
7. ActionExecutionOperator still calls super.close(), and a super.close()
   failure aggregates with component failures rather than replacing them.
8. ResourceCache.close() holds contracts 1, 4, and 5 for a non-Exception
Throwable too: the remaining cached resources, the cache clear, and
   resourceContext.close() all still run.

Contract 1's ordering clause is pinned with InOrder, not left implicit. Order
is load-bearing: resourceCache must close before pythonBridge because cached
resources may hold Python references, PythonActionExecutor.close() calls back
into the interpreter, and super.close() must come last.

Tests

Test	Contract
PythonBridgeManagerTest.closeReleasesInterpreterAndEnvironmentWhenActionExecutorFails	1, 3
PythonBridgeManagerTest.closeReportsFirstFailureWithLaterOnesSuppressed	4
PythonBridgeManagerTest.closeReleasesInterpreterAndEnvironmentWhenActionExecutorThrowsError	5
ActionTaskContextManagerTest.closeClosesContinuationExecutorWhenRunnerContextFails	1, 3
ActionExecutionOperatorTest.closeClosesEveryComponentWhenAnEarlierCloseFails	1, 7
ActionExecutionOperatorTest.closeAggregatesSuperCloseFailureWithComponentFailure	7
ResourceCacheTest.closeClosesEveryResourceWhenAnEarlierResourceThrowsError	8
ResourceCacheTest.closeReportsFirstResourceFailureWithLaterOnesSuppressed	4

Each assertion was checked against a mutation that should break it, rather than
merely observed green. All of these fail the listed tests:

Mutation	Fails
the original close() restored in place	the first five tests, four with Mockito's Wanted but not invoked
IOUtils.closeAll substituted into PythonBridgeManager.close()	only the Error test — the discriminating case for the Throwable decision
ResourceCache catch narrowed back to Exception	closeClosesEveryResourceWhenAnEarlierResourceThrowsError
the super.close() try/catch deleted	both ActionExecutionOperatorTest close tests
if (firstFailure == null) super.close(); — a partial revert of the fix	both ActionExecutionOperatorTest close tests
operator close order swapped (resourceCache ↔ pythonBridge)	both ActionExecutionOperatorTest close tests
PythonBridgeManager close order swapped	closeReleasesInterpreterAndEnvironmentWhenActionExecutorFails

closeReportsFirstResourceFailureWithLaterOnesSuppressed is the one exception to
that: ResourceCache already aggregated Exceptions correctly, so it is a
characterization test guarding that behavior through the rewrite rather than a
regression test for a live defect.

./tools/ut.sh -j passes: 1368 tests, 0 failures, 0 errors (38 skipped, all
pre-existing integration tests that need external services). spotless:check
and RAT are clean.

API

No public API change. All four close() methods keep their
@Override public void close() throws Exception signature.

PythonActionExecutor gains implements AutoCloseable; it already declared a
matching close() throws Exception, so this is additive and no call site
changes.

Two caller-visible behavior changes:

• When several closes fail, the exception received is now the first failure
  rather than the last, and getSuppressed() is non-empty.
• ResourceCache.close() now finishes the remaining closes before propagating a
  non-Exception Throwable, instead of propagating it immediately. The
  Throwable still reaches the caller as itself.

No code in the repo catches these by type, unwraps a cause, or reads
getSuppressed(). ResourceCache.close()'s only production caller is the
ActionExecutionOperator ladder, which catches Throwable.

Documentation

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

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

• Yes
• No

Generated-by: Claude Code (claude-opus-5), also present in the commit messages.

---

Overlap note: #944 rewrites three of these same close() methods with
IOUtils.closeAll and touches a superset of these files. The Error finding is
recorded there so the two do not land opposite decisions on that case;
sequencing is under discussion on that PR.

Same defect class, left out to keep this to one module and one call path, happy
to follow up separately:

• OpenSearchVectorStore.close() — httpClient.close() failing strands
  credentialsProvider
• BedrockEmbeddingModelConnection.close() — embedPool.shutdown() failing
  strands client

[weiqingy]

Thanks for taking this on. A few questions inline.

[weiqingy]

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?

[emecii]

Thanks — I had not connected the two beyond the file overlap, and you are right that the disagreement is on mechanism.

I settled the Error case first, as you suggested, and checked it empirically against flink-core-2.3.0 rather than reading the source. Three closeables that record whether close() ran, first one throwing:

Error     -> a1=true a2=false a3=false     // remaining two never closed
Exception -> b1=true b2=true  b3=true

The chain is closeAll(AutoCloseable...) -> closeAll(Iterable) -> closeAll(Iterable, Exception.class), and the last one rethrows anything not assignable to suppressedException before closing the rest. So closeAll continues past an Exception but stops dead on any non-Exception Throwable, exactly as you read it.

That is now recorded on #944 (#944 (comment)), including the consequence at each of its three call sites — its ActionExecutionOperator.close() passes super::close as the trailing closeable, so an Error from resourceCache skips stateHandler.dispose() too — with a request to adopt the Throwable ladder.

On sequencing, my proposal is this PR first, since it is the smaller change and only rewrites the close() methods: #944's rebase then becomes additive, with longTermMemory and pythonResourceAdapter slotting into ladders that already exist, rather than a mechanism swap in one direction or the other. I have said on #944 that I am equally happy to invert it and rework this as a follow-up on top of theirs. Deferring to whatever you and @joeyutong prefer — the part I care about is that the two do not land opposite decisions on the Error case.

[weiqingy]

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?

[emecii]

Good catch — taken in scope, fixed in 9b5688d.

Not intentional on my part. And I think it is slightly worse than "not a regression": this patch makes that path more consequential rather than less. Before, an Error out of a cached Resource.close() skipped pythonBridge.close() entirely, so nothing was torn down out of order. Now the ladder continues and closes the interpreter while those cached resources are still open — which inverts the very ordering the comment at :556 exists to preserve. Widening ResourceCache restores it, so the two changes belong in the same patch.

ResourceCache.close() now uses the same shape as the other three: Throwable ladders, ExceptionUtils.firstOrSuppressed, ExceptionUtils.rethrowException. Its only production caller is the operator ladder, which already catches Throwable, so nothing downstream changes.

Two tests in ResourceCacheTest. The Error one pins all three consequences you listed, including the two I would otherwise have been assuming from position in the method rather than observing: cache.clear() is checked by reflecting the cache map, and resourceContext.close() by standing a mock SkillManager into the context, since ResourceContextImpl.close() closes it. Narrowing the catch back to Exception fails that test.

I also updated the PR body, which cited ResourceCache.close() as untouched prior art for the aggregation shape — no longer accurate now that this patch modifies it.

[weiqingy]

The test table maps this one to contracts 1 and 7. Contract 1 is well covered by the four verifys, 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 verifys 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?

[emecii]

You are right on both counts, and I confirmed your reasoning by running it: against the code as it stood, deleting the try/catch at :579-583 and replacing it with if (firstFailure == null) super.close(); both left this test green. The second one especially should not have — it is a partial revert of the fix. Contract 7 was not being tested. Fixed in 9b5688d.

Rather than drop it from the table, I think the aggregation half is reachable — you stopped one step short. super.close() does bind statically, but its effect does not: AbstractStreamOperator.close() compiles to stateHandler.dispose() (verified in flink-runtime-2.3.0 bytecode), stateHandler is a protected field, and StreamOperatorStateHandler is a public non-final class with a public non-final dispose(). Swapping the inherited handler therefore makes the super call both observable and failable — no subclass interception needed.

So:

• closeClosesEveryComponentWhenAnEarlierCloseFails now asserts dispose() ran.
• A new closeAggregatesSuperCloseFailureWithComponentFailure makes dispose() throw and asserts the resourceCache failure still reaches the caller with the super failure attached as suppressed — which also pins the firstOrSuppressed argument order, the easy thing to get backwards.

The real handler is restored in the finally before teardown, so the harness still disposes it for real.

Both of your mutations now fail both tests. While in here I also applied your InOrder point from the other thread to these two tests: order is documented even more explicitly on this path (resourceCache before pythonBridge, per the comment at :556), and swapping those two left the tests green until I did. The chain now covers all five components plus stateHandler.dispose() last.

[weiqingy]

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:

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?

[emecii]

Worth it — added in 9b5688d, using your snippet in closeReleasesInterpreterAndEnvironmentWhenActionExecutorFails, with a javadoc note on why the order is load-bearing rather than incidental (the class javadoc at :70-71, and PythonActionExecutor.close() calling back into the interpreter at :205-219). Your swap to {pythonInterpreter, pythonActionExecutor, pythonEnvironmentManager} now fails that test; it passed before.

The clause in contract 1 was meant to carry that weight, so this closes the gap between what it claims and what was checked.

Reviewing the rest of the diff against the same question, ActionExecutionOperator.close() had the identical hole and a more explicitly documented constraint — resourceCache must close before pythonBridge because cached resources may hold Python references (the comment at :556). Swapping those two left both operator close tests green. They now use one InOrder chain across all five components plus stateHandler.dispose(), so the same swap fails there too, and super.close() is pinned last in the same assertion.

[weiqingy]

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?

[weiqingy]

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?