wenjin272 commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3793676469
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -201,6 +201,7 @@ public boolean callPythonAwaitable(String
pythonAwaitableRef) {
return (boolean) ((Object[]) invokeResult)[0];
}
+ @Override
Review Comment:
[P1] `PythonActionExecutor.close()` can still skip the second cleanup step.
If `interpreter.invoke(CLOSE_ASYNC_THREAD_POOL)` throws,
`CLOSE_FLINK_RUNNER_CONTEXT` is never invoked. `PythonBridgeManager` then
continues closing the interpreter, so the runner context's LTM and resource
cache are not explicitly closed first. I reproduced this with a mocked
interpreter: after the first `invoke()` throws, the second invocation is not
made. Could this method use the same first-failure/suppressed-failure ladder
and add a regression test?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/ResourceCache.java:
##########
@@ -140,32 +141,32 @@ public void put(String name, ResourceType type, Resource
resource) {
@Override
public void close() throws Exception {
- Exception firstException = null;
+ // Close every cached resource, then the resource context, even when
an earlier close
+ // fails. The first failure is rethrown with the later ones suppressed.
+ //
+ // The ladders catch Throwable, not Exception:
ActionExecutionOperator.close() closes this
+ // cache before the Python interpreter because cached resources may
hold Python references,
+ // so a non-Exception Throwable escaping here would leave the
remaining resources open
+ // while the interpreter behind them is torn down anyway.
ExceptionUtils.rethrowException
+ // passes Error and Exception through unchanged, so the caller still
sees the original.
+ Throwable firstFailure = null;
for (Map<String, Resource> resources : cache.values()) {
for (Resource resource : resources.values()) {
try {
resource.close();
- } catch (Exception e) {
- if (firstException == null) {
- firstException = e;
- } else {
- firstException.addSuppressed(e);
- }
+ } catch (Throwable t) {
+ firstFailure = ExceptionUtils.firstOrSuppressed(t,
firstFailure);
}
}
}
cache.clear();
try {
resourceContext.close();
- } catch (Exception e) {
- if (firstException == null) {
- firstException = e;
- } else {
- firstException.addSuppressed(e);
- }
+ } catch (Throwable t) {
+ firstFailure = ExceptionUtils.firstOrSuppressed(t, firstFailure);
Review Comment:
[P2] Catching `Throwable` here only aggregates failures after
`ResourceContextImpl.close()` returns or throws, but its nested
[`SkillManager.closeRepos()`](https://github.com/apache/flink-agents/blob/0c4df632101d1a059bb0cb6e1b35cf434cfb5181/runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java#L221-L242)
still catches only `Exception`. If a repository's `close()` throws an `Error`,
the remaining repositories are skipped; `ResourceContextImpl` then clears the
manager reference in `finally`, so that cleanup cannot be retried. The new
close-all guarantee therefore does not reach this nested branch. Could
`SkillManager` use the same `Throwable` aggregation and get a regression test
through the `ResourceCache`/`ResourceContext` path?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/ResourceCache.java:
##########
@@ -140,32 +141,32 @@ public void put(String name, ResourceType type, Resource
resource) {
@Override
public void close() throws Exception {
- Exception firstException = null;
+ // Close every cached resource, then the resource context, even when
an earlier close
+ // fails. The first failure is rethrown with the later ones suppressed.
+ //
+ // The ladders catch Throwable, not Exception:
ActionExecutionOperator.close() closes this
+ // cache before the Python interpreter because cached resources may
hold Python references,
+ // so a non-Exception Throwable escaping here would leave the
remaining resources open
+ // while the interpreter behind them is torn down anyway.
ExceptionUtils.rethrowException
+ // passes Error and Exception through unchanged, so the caller still
sees the original.
+ Throwable firstFailure = null;
for (Map<String, Resource> resources : cache.values()) {
Review Comment:
[P2] The Java path now closes every resource after an earlier failure, but
the equivalent [Python
`ResourceCache.close()`](https://github.com/apache/flink-agents/blob/0c4df632101d1a059bb0cb6e1b35cf434cfb5181/python/flink_agents/runtime/resource_cache.py#L96-L108)
is still sequential: the first resource failure skips the remaining resources,
`_cache.clear()`, and `_resource_context.close()`. I reproduced
`first_closed=True, second_closed=False, cache_size=1`. Could we align the
Python lifecycle behavior and add the corresponding regression test?
--
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]
