[FLINK-21072][hotfix] Clear resources in HeapSnapshotStrategy

zobar · Jan 28, 2021 · 6cc2c72 · 6cc2c72
1 parent 2596c12
commit 6cc2c72
Showing 1 changed file with 8 additions and 4 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapSnapshotStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapSnapshotStrategy.java
@@ -247,16 +247,20 @@ static class HeapSnapshotResources implements SnapshotResources {
         private final Map<StateUID, Integer> stateNamesToId;
 
         HeapSnapshotResources(
-                List<StateMetaInfoSnapshot> metaInfoSnapshots,
-                Map<StateUID, StateSnapshot> cowStateStableSnapshots,
-                Map<StateUID, Integer> stateNamesToId) {
+                @Nonnull List<StateMetaInfoSnapshot> metaInfoSnapshots,
+                @Nonnull Map<StateUID, StateSnapshot> cowStateStableSnapshots,
+                @Nonnull Map<StateUID, Integer> stateNamesToId) {
             this.metaInfoSnapshots = metaInfoSnapshots;
             this.cowStateStableSnapshots = cowStateStableSnapshots;
             this.stateNamesToId = stateNamesToId;
         }
 
         @Override
-        public void release() {}
+        public void release() {
+            for (StateSnapshot stateSnapshot : cowStateStableSnapshots.values()) {
+                stateSnapshot.release();
+            }
+        }
 
         public List<StateMetaInfoSnapshot> getMetaInfoSnapshots() {
             return metaInfoSnapshots;