[FLINK-23003][runtime] Fix resource leak in RocksIncrementalSnapshotStrategy

Author: leiyanfei

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java

@@ -465,6 +465,7 @@ public void dispose() {
 
             cleanInstanceBasePath();
         }
+        IOUtils.closeQuietly(checkpointSnapshotStrategy);
         this.disposed = true;
     }
 

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java

@@ -265,7 +265,7 @@ public RocksDBKeyedStateBackend<K> build() throws BackendBuildingException {
                 new RocksDbTtlCompactFiltersManager(ttlTimeProvider);
 
         ResourceGuard rocksDBResourceGuard = new ResourceGuard();
-        RocksDBSnapshotStrategyBase<K, ?> checkpointStrategy;
+        RocksDBSnapshotStrategyBase<K, ?> checkpointStrategy = null;
         PriorityQueueSetFactory priorityQueueFactory;
         SerializedCompositeKeyBuilder<K> sharedRocksKeyBuilder;
         // Number of bytes required to prefix the key groups.
@@ -363,6 +363,7 @@ public RocksDBKeyedStateBackend<K> build() throws BackendBuildingException {
             IOUtils.closeQuietly(optionsContainer);
             ttlCompactFiltersManager.disposeAndClearRegisteredCompactionFactories();
             kvStateInformation.clear();
+            IOUtils.closeQuietly(checkpointStrategy);
             try {
                 FileUtils.deleteDirectory(instanceBasePath);
             } catch (Exception ex) {

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/snapshot/RocksDBSnapshotStrategyBase.java

@@ -43,7 +43,7 @@
  * @param <K> type of the backend keys.
  */
 public abstract class RocksDBSnapshotStrategyBase<K, R extends SnapshotResources>
-        implements CheckpointListener, SnapshotStrategy<KeyedStateHandle, R> {
+        implements CheckpointListener, SnapshotStrategy<KeyedStateHandle, R>, AutoCloseable {
 
     private static final Logger LOG = LoggerFactory.getLogger(RocksDBSnapshotStrategyBase.class);
 
@@ -92,4 +92,7 @@ public RocksDBSnapshotStrategyBase(
     public String getDescription() {
         return description;
     }
+
+    @Override
+    public abstract void close();
 }

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/snapshot/RocksFullSnapshotStrategy.java

@@ -141,6 +141,11 @@ public void notifyCheckpointAborted(long checkpointId) {
         // nothing to do.
     }
 
+    @Override
+    public void close() {
+        // nothing to do.
+    }
+
     private SupplierWithException<CheckpointStreamWithResultProvider, Exception>
             createCheckpointStreamSupplier(
                     long checkpointId,

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/snapshot/RocksIncrementalSnapshotStrategy.java

@@ -204,6 +204,11 @@ public void notifyCheckpointAborted(long abortedCheckpointId) {
         }
     }
 
+    @Override
+    public void close() {
+        stateUploader.close();
+    }
+
     @Nonnull
     private SnapshotDirectory prepareLocalSnapshotDirectory(long checkpointId) throws IOException {