(flink) 01/02: [FLINK-40921][runtime] Invoke statebackend's close() and dispose() in right order

[zakelly]

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 <zakelly....@gmail.com>
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);
+            }
         }
     }