[FLINK-26757][state] Change the default value of state.backend.rocksd…

…b.restore-overlap-fraction-threshold to 0
lightClouds917 · Apr 12, 2022 · 4034d3c · 4034d3c
1 parent 87dd6fb
commit 4034d3c
Show file tree

Hide file tree

Showing 5 changed files with 32 additions and 13 deletions.
diff --git a/docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html b/docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html
@@ -94,9 +94,9 @@
         </tr>
         <tr>
             <td><h5>state.backend.rocksdb.restore-overlap-fraction-threshold</h5></td>
-            <td style="word-wrap: break-word;">0.75</td>
+            <td style="word-wrap: break-word;">0.0</td>
             <td>Double</td>
-            <td>The threshold of the overlap fraction between the handle's key-group range and target key-group range. When restore base DB, only the handle which overlap fraction greater than or equal to *threshold* has a chance to be an initial handle.</td>
+            <td>The threshold of overlap fraction between the handle's key-group range and target key-group range. When restore base DB, only the handle which overlap fraction greater than or equal to threshold has a chance to be an initial handle. The default value is 0.0, there is always a handle will be selected for initialization. </td>
         </tr>
         <tr>
             <td><h5>state.backend.rocksdb.thread.num</h5></td>

diff --git a/...b/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java b/...b/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java
@@ -498,7 +498,7 @@ public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
                         .setNativeMetricOptions(
                                 resourceContainer.getMemoryWatcherOptions(defaultMetricOptions))
                         .setWriteBatchSize(getWriteBatchSize())
-                        .setOverlapFractionThreshold(overlapFractionThreshold);
+                        .setOverlapFractionThreshold(getOverlapFractionThreshold());
         return builder.build();
     }
 
@@ -833,6 +833,12 @@ public void setWriteBatchSize(long writeBatchSize) {
         this.writeBatchSize = writeBatchSize;
     }
 
+    double getOverlapFractionThreshold() {
+        return overlapFractionThreshold == UNDEFINED_OVERLAP_FRACTION_THRESHOLD
+                ? RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue()
+                : overlapFractionThreshold;
+    }
+
     // ------------------------------------------------------------------------
     //  utilities
     // ------------------------------------------------------------------------

diff --git a/...db/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBConfigurableOptions.java b/...db/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBConfigurableOptions.java
@@ -258,11 +258,12 @@ public class RocksDBConfigurableOptions implements Serializable {
     public static final ConfigOption<Double> RESTORE_OVERLAP_FRACTION_THRESHOLD =
             key("state.backend.rocksdb.restore-overlap-fraction-threshold")
                     .doubleType()
-                    .defaultValue(0.75)
+                    .defaultValue(0.0)
                     .withDescription(
-                            "The threshold of the overlap fraction between the handle's key-group range and target key-group range. "
-                                    + "When restore base DB, only the handle which overlap fraction greater than or equal to *threshold* "
-                                    + "has a chance to be an initial handle.");
+                            "The threshold of overlap fraction between the handle's key-group range and target key-group range. "
+                                    + "When restore base DB, only the handle which overlap fraction greater than or equal to threshold "
+                                    + "has a chance to be an initial handle. "
+                                    + "The default value is 0.0, there is always a handle will be selected for initialization. ");
 
     static final ConfigOption<?>[] CANDIDATE_CONFIGS =
             new ConfigOption<?>[] {

diff --git a/.../java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtilsTest.java b/.../java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtilsTest.java
@@ -96,12 +96,6 @@ public void testChooseTheBestStateHandleForInitial() {
         when(keyedStateHandle3.getKeyGroupRange()).thenReturn(new KeyGroupRange(8, 12));
         keyedStateHandles.add(keyedStateHandle3);
 
-        Assert.assertNull(
-                RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
-                        keyedStateHandles,
-                        new KeyGroupRange(3, 5),
-                        RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue()));
-
         // this should choose keyedStateHandle2, because keyedStateHandle2's key-group range
         // satisfies the overlap fraction demand.
         Assert.assertEquals(

diff --git a/...src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java b/...src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
@@ -771,6 +771,24 @@ public void testConfigureIllegalMemoryControlParameters() {
         }
     }
 
+    @Test
+    public void testDefaultRestoreOverlapThreshold() {
+        EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
+        assertTrue(
+                RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD.defaultValue()
+                        == rocksDBStateBackend.getOverlapFractionThreshold());
+    }
+
+    @Test
+    public void testConfigureRestoreOverlapThreshold() {
+        EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
+        Configuration configuration = new Configuration();
+        configuration.setDouble(RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD, 0.3);
+        rocksDBStateBackend =
+                rocksDBStateBackend.configure(configuration, getClass().getClassLoader());
+        assertTrue(0.3 == rocksDBStateBackend.getOverlapFractionThreshold());
+    }
+
     private void verifySetParameter(Runnable setter) {
         try {
             setter.run();