[FLINK-38558] Require copy when use heap timer

Author: Zakelly

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackend.java

@@ -938,7 +938,7 @@ public boolean requiresLegacySynchronousTimerSnapshots(SnapshotType checkpointTy
 
     @Override
     public boolean isSafeToReuseKVState() {
-        return true;
+        return !(priorityQueueFactory instanceof HeapPriorityQueueSetFactory);
     }
 
     @VisibleForTesting

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendTest.java

@@ -51,6 +51,7 @@
 
 import static org.apache.flink.state.forst.ForStConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
 import static org.apache.flink.state.forst.ForStOptions.LOCAL_DIRECTORIES;
+import static org.apache.flink.state.forst.ForStOptions.TIMER_SERVICE_FACTORY;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for the partitioned state part of {@link ForStStateBackendTest}. */
@@ -64,7 +65,8 @@ public static List<Object[]> modes() {
                 new Object[][] {
                     {
                         (SupplierWithException<CheckpointStorage, IOException>)
-                                JobManagerCheckpointStorage::new
+                                JobManagerCheckpointStorage::new,
+                        false
                     },
                     {
                         (SupplierWithException<CheckpointStorage, IOException>)
@@ -73,12 +75,17 @@ public static List<Object[]> modes() {
                                             TempDirUtils.newFolder(tempFolder).toURI().toString();
                                     return new FileSystemCheckpointStorage(
                                             new Path(checkpointPath), 0, -1);
-                                }
+                                },
+                        true
                     }
                 });
     }
 
-    @Parameter public SupplierWithException<CheckpointStorage, IOException> storageSupplier;
+    @Parameter(value = 0)
+    public SupplierWithException<CheckpointStorage, IOException> storageSupplier;
+
+    @Parameter(value = 1)
+    public boolean useHeapTimer;
 
     @Override
     protected CheckpointStorage getCheckpointStorage() throws Exception {
@@ -91,6 +98,11 @@ protected ConfigurableStateBackend getStateBackend() throws Exception {
         Configuration config = new Configuration();
         config.set(LOCAL_DIRECTORIES, tempFolder.toString());
         config.set(USE_INGEST_DB_RESTORE_MODE, true);
+        config.set(
+                TIMER_SERVICE_FACTORY,
+                useHeapTimer
+                        ? ForStStateBackend.PriorityQueueStateType.HEAP
+                        : ForStStateBackend.PriorityQueueStateType.ForStDB);
         return backend.configure(config, Thread.currentThread().getContextClassLoader());
     }
 
@@ -109,7 +121,7 @@ protected boolean supportsAsynchronousSnapshots() {
      */
     @Override
     protected boolean isSafeToReuseKVState() {
-        return true;
+        return !useHeapTimer;
     }
 
     @TestTemplate

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

@@ -1108,7 +1108,7 @@ public boolean requiresLegacySynchronousTimerSnapshots(SnapshotType checkpointTy
 
     @Override
     public boolean isSafeToReuseKVState() {
-        return true;
+        return !(priorityQueueFactory instanceof HeapPriorityQueueSetFactory);
     }
 
     @Override

flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/EmbeddedRocksDBStateBackendTest.java

@@ -99,6 +99,7 @@
 import static org.apache.flink.state.rocksdb.RocksDBConfigurableOptions.INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE;
 import static org.apache.flink.state.rocksdb.RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING;
 import static org.apache.flink.state.rocksdb.RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
+import static org.apache.flink.state.rocksdb.RocksDBOptions.TIMER_SERVICE_FACTORY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -131,13 +132,15 @@ public static List<Object[]> modes() {
                         true,
                         (SupplierWithException<CheckpointStorage, IOException>)
                                 JobManagerCheckpointStorage::new,
+                        false,
                         false
                     },
                     {
                         true,
                         (SupplierWithException<CheckpointStorage, IOException>)
                                 JobManagerCheckpointStorage::new,
-                        true
+                        true,
+                        false
                     },
                     {
                         false,
@@ -148,7 +151,8 @@ public static List<Object[]> modes() {
                                     return new FileSystemCheckpointStorage(
                                             new Path(checkpointPath), 0, -1);
                                 },
-                        false
+                        false,
+                        true
                     }
                 });
     }
@@ -162,6 +166,9 @@ public static List<Object[]> modes() {
     @Parameter(value = 2)
     public boolean useIngestDB;
 
+    @Parameter(value = 3)
+    public boolean useHeapTimer;
+
     // Store it because we need it for the cleanup test.
     private String dbPath;
     private RocksDB db = null;
@@ -203,8 +210,10 @@ private Configuration createBackendConfig() {
         Configuration configuration = new Configuration();
         configuration.set(USE_INGEST_DB_RESTORE_MODE, useIngestDB);
         configuration.set(
-                RocksDBOptions.TIMER_SERVICE_FACTORY,
-                EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB);
+                TIMER_SERVICE_FACTORY,
+                useHeapTimer
+                        ? EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP
+                        : EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB);
         configuration.set(RocksDBManualCompactionOptions.MIN_INTERVAL, Duration.ofMillis(1));
         return configuration;
     }
@@ -234,7 +243,7 @@ protected boolean supportsAsynchronousSnapshots() {
 
     @Override
     protected boolean isSafeToReuseKVState() {
-        return true;
+        return !useHeapTimer;
     }
 
     // small safety net for instance cleanups, so that no native objects are left