This is an automated email from the ASF dual-hosted git repository. Zakelly pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8407b6e993b92d4eca3ed98c067dbc4297b7359e Author: Zakelly <[email protected]> AuthorDate: Tue Oct 6 12:47:42 2026 +0800 [FLINK-40921][runtime] Invoke statebackend's close() and dispose() in right order --- .../api/operators/StreamOperatorStateHandler.java | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java index 09dcd710ee7..7a9b11b5b4e 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java @@ -154,15 +154,8 @@ public class StreamOperatorStateHandler { public void dispose() throws Exception { try (Closer closer = Closer.create()) { - if (closeableRegistry.unregisterCloseable(operatorStateBackend)) { - closer.register(operatorStateBackend); - } - if (closeableRegistry.unregisterCloseable(keyedStateBackend)) { - closer.register(keyedStateBackend); - } - if (closeableRegistry.unregisterCloseable(asyncKeyedStateBackend)) { - closer.register(asyncKeyedStateBackend); - } + // The guava's closer will invoke these registered methods in reverse order of + // registration, so we register dispose before close here. if (operatorStateBackend != null) { closer.register(operatorStateBackend::dispose); } @@ -172,6 +165,15 @@ public class StreamOperatorStateHandler { if (asyncKeyedStateBackend != null) { closer.register(asyncKeyedStateBackend::dispose); } + if (closeableRegistry.unregisterCloseable(operatorStateBackend)) { + closer.register(operatorStateBackend); + } + if (closeableRegistry.unregisterCloseable(keyedStateBackend)) { + closer.register(keyedStateBackend); + } + if (closeableRegistry.unregisterCloseable(asyncKeyedStateBackend)) { + closer.register(asyncKeyedStateBackend); + } } }
