weiqingy opened a new pull request, #974: URL: https://github.com/apache/flink-agents/pull/974
Linked issue: none, this is a hotfix. ### Purpose of change `FlussActionStateStore.close()` wraps `table.close()` in a `try` with `connection.close()` in the `finally`. That does close the connection when the table close throws, so there is no leak. But a `finally` block that throws discards the in-flight exception, and two failures were being lost because of it. When both closes throw, the caller only sees the connection failure and gets no signal that the table close failed at all. When `table.close()` throws an `Error`, a connection failure from the `finally` replaces it, so the `Error` never reaches the caller. An OOM or a `LinkageError` during table close would surface as an ordinary Fluss connection-close failure. The clearest argument for the fix is internal to the class. Its constructor's cleanup path already attaches every cleanup failure to the original exception before rethrowing (`FlussActionStateStore.java:170-184`), so `close()` was contradicting its own constructor twenty lines up. `close()` now records the table failure, still closes the connection in the `finally`, and attaches a connection failure as a suppressed exception instead of letting it replace the original. One design note, since the obvious question is why this does not look like the `firstException` ladder in `KafkaActionStateStore.close()`. The `finally` is kept deliberately rather than replaced by two flat `catch (Exception)` arms. A `finally` runs for any `Throwable` and `catch (Exception)` does not, so a flat ladder would skip the connection close when `table.close()` throws an `Error`, and leak the connection. The two classes are not symmetric here: `KafkaActionStateStore.close()` had no `finally` before its own fix, so a flat ladder narrowed nothing there, whereas this class does have one to preserve. `IOUtils.closeAll` was considered and does not fit. Its first-wins suppression and null-skipping match this contract, but with the default `Exception.class` it rethrows a non-`Exception` `Throwable` immediately without closing the remaining resources, which is the same leak. Passing `Throwable.class` avoids that but wraps the throwable in `new Exception(e)`, so an `OutOfMemoryError` would reach the caller as an ordinary `Exception`. ### Tests Five tests in `FlussActionStateStoreTest`, one per outcome: both closes succeed, the table close fails, both fail, only the connection close fails, and the table close throws a non-`Exception` `Throwable`. Two of them fail against the previous `close()`, which is what pins the two defects described above: `testCloseKeepsTableFailureWhenBothCloseFail` and `testCloseKeepsTableErrorWhenConnectionCloseAlsoFails`. The other three pass both before and after, as regression pins. The stubs deliberately span checked (`IOException`), unchecked (`RuntimeException`) and `Error` failures. With unchecked stubs only, narrowing either `catch (Exception e)` to `catch (RuntimeException e)` passes the whole suite while reinstating the original defect for checked exceptions, which is the likely case in practice given that `FlussConnection.close()` calls `RemoteFileDownloader.close() throws IOException`. The last test asserts only that the throwable stays the primary failure and that the connection is still closed. It deliberately says nothing about the connection exception that is dropped while an `Error` is in flight, so it stays valid if the method ever widens to `catch (Throwable)`. `./tools/ut.sh -j` passes: 1085 tests, 0 failures, 0 errors. The `dist` modules build for Flink 1.20, 2.0, 2.1, 2.2 and 2.3. ### API No API change. `close()` keeps its `@Override public void close() throws Exception` signature. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No `Generated-by: Claude Code (claude-opus-5)`, also present in the commit message. -- 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]
