emecii commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3742785700
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
@Override
public void close() throws Exception {
- if (pythonActionExecutor != null) {
- pythonActionExecutor.close();
- }
- if (pythonInterpreter != null) {
- pythonInterpreter.close();
- }
- if (pythonEnvironmentManager != null) {
- pythonEnvironmentManager.close();
- }
+ IOUtils.closeAll(
Review Comment:
Following up on @weiqingy's note in #987 that these two PRs disagree on
mechanism rather than merely touching the same seven files. I'm the author of
#987, so flagging my own interest up front.
I checked the `Error` case empirically against `flink-core-2.3.0` rather
than reading the source, and the concern holds. Three closeables that record
whether `close()` ran, the first one throwing:
```
A (first throws OutOfMemoryError): thrown=java.lang.OutOfMemoryError: boom
A closed flags -> a1=true a2=false a3=false
B (first throws IllegalStateException):
thrown=java.lang.IllegalStateException: boom
B closed flags -> b1=true b2=true b3=true
```
The varargs overload delegates to `closeAll(Iterable)`, which delegates to
`closeAll(Iterable, Class<T>)` with `suppressedException = Exception.class`;
that method rethrows anything not assignable to it *before* closing the rest.
So `closeAll` continues past an `Exception` but stops dead on any
non-`Exception` `Throwable`.
At this call site that means an `Error` out of `longTermMemory.close()` or
`pythonActionExecutor.close()` leaves `pythonInterpreter` and
`pythonEnvironmentManager` unclosed — the native Python state this PR exists to
release, retained for the lifetime of the TaskManager JVM. The same shape
appears at the other two sites:
- `ActionExecutionOperator.close()`: an `Error` from `resourceCache` skips
`contextManager`, `pythonBridge`, `eventRouter`, `durableExecManager`, and the
trailing `super::close`, so `stateHandler.dispose()` is skipped too.
- `ActionTaskContextManager.close()`: an `Error` from the runner context
strands the continuation executor's thread pool.
Worth noting that `ResourceCache.close()` (`ResourceCache.java:148` and
`:160`) currently catches only `Exception`, so an `Error` out of a cached
`Resource.close()` propagates unchanged and is a concrete way to reach the
first of those.
Would you consider replacing `closeAll` with a `catch (Throwable)` ladder
that aggregates via `ExceptionUtils.firstOrSuppressed` and rethrows via
`ExceptionUtils.rethrowException`? It keeps the same
first-failure-wins-with-later-ones-suppressed semantics, and `rethrowException`
passes both `Error` and `Exception` through unwrapped, so callers still see the
original type and instance. That is the shape `KafkaActionStateStore.close()`
(#948) and `ResourceCache.close()` already use in this module.
On sequencing, since we overlap on seven files: #987 is the smaller change
and only rewrites the three `close()` methods. If it lands first, this PR's
rebase becomes additive — `longTermMemory` and `pythonResourceAdapter` slot
into ladders that already exist — instead of a mechanism swap in one direction
or the other. I'm equally happy to go the other way and rework #987 as a
follow-up on top of this if you'd prefer not to reshuffle. Mainly I'd like the
two not to land opposite decisions on the `Error` case. What works best for you?
--
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]
