emecii commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3793775972
##########
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:
Agreed, and fixed in 23312147. Your reproduction matches what I see: the two
cleanups were sequential statements, so a throwing `CLOSE_ASYNC_THREAD_POOL`
skipped `CLOSE_FLINK_RUNNER_CONTEXT` entirely.
It's the worst place in the chain for that, for the reason you give —
`PythonBridgeManager` closes the interpreter on the very next rung, so the
skipped cleanup has no later chance to run. `close()` now uses the same
first-failure/suppressed ladder as the managers above it, and still nulls
`pythonRunnerContext` in the `finally` so a repeated close can't double-free.
Three tests in `PythonActionExecutorTest`: the runner-context cleanup still
runs when the shutdown fails, the first failure is rethrown with the later one
suppressed, and the `Error` case. Restoring the original method fails all three.
One note on scope, since this overlaps #944: @joeyutong had offered to
handle this method's `Error` path as part of their rebase, which is why I'd
left it alone. Doing it here instead means that's one less thing for the
rebase, and their `IOUtils.closeAll` inside this method lands on top of a
ladder rather than replacing one. Flagging so the two of you aren't both
holding it.
##########
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:
Good catch — this is the branch the guarantee genuinely did not reach, and
the `finally` you point at is what makes it unrecoverable rather than merely
delayed. Fixed in 23312147.
`closeRepos()` now uses the `Throwable` ladder with
`firstOrSuppressed`/`rethrowException`. Two tests, including the one through
the path you asked for:
`ResourceCacheTest.closeClosesEverySkillRepositoryWhenAnEarlierRepoThrowsError`
drives real `SkillRepository` instances through `ResourceCache.close()` →
`ResourceContextImpl.close()` → `SkillManager.close()`, rather than asserting
at the `SkillManager` level only.
Worth reporting how the first attempt went wrong, since it affects how much
the result is worth. I initially wrote both tests with one failing repo and one
healthy one — and the mutation check showed the `SkillManager` test *passing*
against the narrowed catch. The de-dup set is an `IdentityHashMap`, so
iteration order is unspecified: when the healthy repo happens to run first,
both end up closed even with the bug present. Both tests now use two failing
repos, which fails deterministically whichever runs first; I confirmed that
over three runs before pushing.
Also updated a comment in
`errorDuringRegistrationStillClosesRepoAndSuppressesCloseError` that documented
the old limitation ("closeRepos() catches only Exception per repo, so an Error
ends the iteration") as intended behavior — no longer true.
--
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]
