emecii opened a new pull request, #987:
URL: https://github.com/apache/flink-agents/pull/987
Linked issue: none (hotfix)
### Purpose of change
Three `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 |
#### 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.
#### 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()`,
which
already aggregate this way in the same module.
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. An earlier revision of this
patch
used `closeAll` and the `Error` test below is what caught it.
`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.
### Tests
Five tests, each verified against the pre-fix code rather than merely
observed
green.
| Test | Contract | Fails against |
|---|---|---|
|
`PythonBridgeManagerTest.closeReleasesInterpreterAndEnvironmentWhenActionExecutorFails`
| 1, 3 | original `close()` |
| `PythonBridgeManagerTest.closeReportsFirstFailureWithLaterOnesSuppressed`
| 4 | original `close()` |
|
`PythonBridgeManagerTest.closeReleasesInterpreterAndEnvironmentWhenActionExecutorThrowsError`
| 5 | original `close()`, and the `IOUtils.closeAll` revision |
|
`ActionTaskContextManagerTest.closeClosesContinuationExecutorWhenRunnerContextFails`
| 1, 3 | original `close()` |
|
`ActionExecutionOperatorTest.closeClosesEveryComponentWhenAnEarlierCloseFails`
| 1, 7 | original `close()` |
Each was run against the original `close()` restored in place; all five fail
there, four with Mockito's `Wanted but not invoked` — that is, the component
behind the failing one is genuinely never closed today.
The `Error` test is the discriminating one for the `Throwable` decision.
Substituting `IOUtils.closeAll` back into `PythonBridgeManager.close()`
leaves
the other three tests green and fails only that one.
`./tools/ut.sh -j` passes: 1365 tests, 0 failures, 0 errors (38 skipped, all
pre-existing integration tests that need external services).
`spotless:check` is
clean.
### API
No public API change. All three `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.
One caller-visible behavior change: when several closes fail, the exception
received is now the first failure rather than the last, and
`getSuppressed()` is
non-empty. No code in the repo catches these by type, unwraps a cause, or
reads
`getSuppressed()`.
### Documentation
- [ ] `doc-needed`
- [x] `doc-not-needed`
- [ ] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
`Generated-by: Claude Code (claude-opus-5)`, also present in the commit
message.
---
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`
--
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]
