wenjin272 commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3793676473
##########
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] Java 侧修复了 stop-on-first-failure，但等价的 [Python
`ResourceCache.close()`](https://github.com/apache/flink-agents/blob/0c4df632101d1a059bb0cb6e1b35cf434cfb5181/python/flink_agents/runtime/resource_cache.py#L96-L108)
仍是串行关闭：首个资源失败会跳过后续资源、`_cache.clear()` 和 `_resource_context.close()`。复现结果是
`first_closed=True, second_closed=False, cache_size=1`。Java/Python 的
resource-cache 生命周期应保持一致，建议同步修改 Python 实现并补对应回归测试。
##########
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] 这里把 executor 纳入统一关闭链路后，`close()` 自身仍可能跳过第二个清理动作：如果
`CLOSE_ASYNC_THREAD_POOL` 的 `interpreter.invoke()` 抛出异常，后面的
`CLOSE_FLINK_RUNNER_CONTEXT` 就不会执行。随后 `PythonBridgeManager` 会继续关闭
interpreter，runner context 中的 LTM/resource cache 因而没有先被显式释放。我用 mock interpreter
复现过：第一次 `invoke()` 抛出后，第二次调用不会发生。建议这里也采用 first-failure/suppressed-failure
聚合，并补一个回归测试。
##########
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] 这里捕获 `Throwable` 只能保证 `ResourceContextImpl.close()` 返回或抛出后继续聚合，但它内部的
[`SkillManager.closeRepos()`](https://github.com/apache/flink-agents/blob/0c4df632101d1a059bb0cb6e1b35cf434cfb5181/runtime/src/main/java/org/apache/flink/agents/runtime/skill/SkillManager.java#L221-L242)
仍然只捕获 `Exception`。如果某个 repository 的 `close()` 抛出 `Error`，后续 repositories
会被跳过；随后 `ResourceContextImpl` 又会在 `finally` 中清掉 manager 引用，导致这部分清理无法重试。因此这里的
close-all 保证还没有覆盖嵌套分支。建议 `SkillManager` 也采用相同的 `Throwable` 聚合，并通过
ResourceCache/ResourceContext 路径补回归测试。
--
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]
