CodeTrainerMan opened a new pull request, #1221: URL: https://github.com/apache/flink-agents/pull/1221
Linked issue: #1108 ### Purpose of change A caller can no longer lose a repository cleanup failure. `SkillManager.close()` attempted every repository but swallowed whatever `SkillRepository.close()` raised: the failure was neither re-raised nor logged, so an extension-provided repository that failed to release its temp directory failed silently. It now attempts every repository and re-raises the first failure, matching the Java `SkillManager.closeRepos()` contract and the Python `ResourceCache.close()` pattern. The gap is a parity defect on a public extension point rather than a live incident: the in-tree repository implementations do not raise on close. #### Runtime flow `close()` walks `_opened_repos` in open order, skipping a repository whose identity was already seen, and calls `failure_of(repo.close)` for each. `first_or_logged` keeps the first failure and logs any later one at WARNING; after the loop the first failure is re-raised unchanged. `_opened_repos` is walked rather than `_repos.values()`, because a duplicate skill name displaces the earlier repository in `_repos` while that repository is still owned and must be closed. On the load path, `SkillManager.__init__` calls `_load_skills()`, which on failure closes what it opened, logs a failure from that cleanup, and re-raises the original load failure. `failure_of` and `first_or_logged` now live in `runtime/close_utils.py`. They are unchanged; `resource_cache` and `flink_runner_context` import them from there. #### Key decisions - Extracted rather than duplicated. `skill_manager` cannot import them from `resource_cache` at module level: `skill_manager -> resource_cache -> resource_context -> skill_manager` is an import cycle. A function-local import would have worked at call time and was rejected, because it leaves two copies of one contract to keep in step, and the point of the change is that this contract matches `ResourceCache.close()`. - `Exception`, not `BaseException`, matching `ResourceCache.close()`: cleanup must not absorb `KeyboardInterrupt` or `SystemExit`. The cost is that a `BaseException` from one repository ends the loop early where Java's `Throwable` catch would continue. - `_opened_repos` is not cleared, matching Java. Idempotency rests on the repository implementations, so one that fails does so again on a second call. - A cleanup failure on the load path is logged, not attached: Python 3.10 has no `addSuppressed` and no `ExceptionGroup`, and the load failure is what the caller needs. ### Behavioral Semantics #### Interaction decisions | Repository `close()` | Where | Result | |---|---|---| | raises `Exception`, first to fail | `close()` | later repositories still attempted; that failure re-raised after the loop | | raises `Exception`, not first | `close()` | logged at WARNING; does not replace the first | | raises `BaseException` | `close()` | propagates immediately; remaining repositories not attempted | | raises `Exception` | cleanup during a failed load | logged; the load failure is still what propagates | | does not raise | `close()` | returns `None`; every repository closed | | one repository registered twice | either | closed once, deduped by identity | #### Behavioral contracts 1. `close()` attempts every distinct opened repository even when an earlier one raises. 2. The first failure is re-raised after the loop, as the same exception object, not wrapped. 3. Later failures are logged and never replace the first. 4. When nothing fails, `close()` returns `None` and closes every repository. 5. A failed load still raises the load failure; a failure while closing what was opened is logged instead. 6. `ResourceCache.close()` and `FlinkRunnerContext.close()` behave as before, including which failure they re-raise. #### Failure behavior - Repository `close()` raising `Exception`: attempted, then kept or logged per contracts 1-3. Nothing retries and nothing falls back. - Repository `close()` raising `BaseException`: propagates unwrapped, and repositories behind it are not attempted. Deliberate, matching `ResourceCache.close()`. - Failure in the cleanup that runs when a load fails: logged at WARNING with the exception attached; the original failure propagates. - `close()` raising inside `__exit__`: propagates, chained as the context of any exception already in flight. - Invalid configuration and external-service errors are load-path concerns, unchanged here. ### Tests `python/flink_agents/runtime/skill/tests/test_manager_close.py`, new, five cases, using a test repository whose `close()` records the call and can be made to raise. Each fails against the implementation that swallowed failures. | Contract | Tests | |---|---| | 1, every repository attempted | `test_close_attempts_every_repo_and_rethrows_first_failure` | | 2, first failure re-raised unchanged | `test_close_attempts_every_repo_and_rethrows_first_failure`, `test_close_reports_first_failure_when_several_fail` | | 3, later failures logged | `test_close_logs_later_failures` | | 4, healthy path | `test_close_returns_normally_when_nothing_fails` | | 5, load failure survives cleanup | `test_load_failure_is_not_replaced_by_close_failure` | | 6, other callers unchanged | no test; the helpers are moved verbatim and only the import sites change | Coverage by risk: the surfaced-failure paths carry the risk and are pinned. Not verified: no test exercises a `BaseException` from `close()`; no test drives a real URL or package repository through a failing close, only a test double; the suites that import PyFlink, among them `test_resource_cache_close.py`, were not run locally, so contract 6 rests on the CI run. <details> <summary>Implementation invariants and supporting evidence</summary> - Repositories are deduped by `id()`, so one contributing two skills is closed once; iteration is open order. - The log message is `Suppressed failure closing skill repository.`, produced by the shared helper, so it reads the same as the `ResourceCache` and `FlinkRunnerContext` messages. - `failure_of` and `first_or_logged` are the functions `resource_cache.py` had, unchanged; `flink_runner_context.py` was repointed to the new module. - `ResourceCache.close()` keeps its docstring, which already states the `Exception`-not-`BaseException` rationale. </details> ### API No signature change and no new public class. `close_utils` is runtime-internal. `SkillRepository.close()` is a public extension point, and this is the compatibility cost: an implementation that raises now makes `SkillManager.close()` raise where it previously returned `None`, and `ResourceContextImpl.close()` propagates it into `ResourceCache.close()`, which keeps it as the first failure. That is the intent: a repository that cannot release its temp directory should say so on operator close, including Flink failover. The bundled filesystem, URL, package and classpath repositories do not raise, and `ResourceCache.close()` behavior is unchanged, so a caller using only those sees no change. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes Generated-by: CodeBuddy (Auto) -- 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]
