[FLINK-23789][rocksdb] Remove unnecessary setTotalOrderForSeek for Rocks iterator

This fix apache#16834.

Author: Myasuka

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

@@ -127,7 +127,7 @@ private static void deleteRange(
             throws RocksDBException {
 
         for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
-            try (ReadOptions readOptions = RocksDBOperationUtils.createTotalOrderSeekReadOptions();
+            try (ReadOptions readOptions = new ReadOptions();
                     RocksIteratorWrapper iteratorWrapper =
                             RocksDBOperationUtils.getRocksIterator(
                                     db, columnFamilyHandle, readOptions);

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

@@ -104,16 +104,6 @@ public static RocksIteratorWrapper getRocksIterator(
         return new RocksIteratorWrapper(db.newIterator(columnFamilyHandle, readOptions));
     }
 
-    /**
-     * Create a total order read option to avoid user misuse, see FLINK-17800 for more details.
-     *
-     * <p>Note, remember to close the generated {@link ReadOptions} when dispose.
-     */
-    // TODO We would remove this method once we bump RocksDB version larger than 6.2.2.
-    public static ReadOptions createTotalOrderSeekReadOptions() {
-        return new ReadOptions().setTotalOrderSeek(true);
-    }
-
     public static void registerKvStateInformation(
             Map<String, RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation,
             RocksDBNativeMetricMonitor nativeMetricMonitor,

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

@@ -183,9 +183,7 @@ public WriteOptions getWriteOptions() {
 
     /** Gets the RocksDB {@link ReadOptions} to be used for read operations. */
     public ReadOptions getReadOptions() {
-        // We ensure total order seek by default to prevent user misuse, see FLINK-17800 for more
-        // details
-        ReadOptions opt = RocksDBOperationUtils.createTotalOrderSeekReadOptions();
+        ReadOptions opt = new ReadOptions();
         handlesToClose.add(opt);
 
         // add user-defined options factory, if specified

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

@@ -422,7 +422,7 @@ private RestoredDBInstance(
             this.columnFamilyHandles = columnFamilyHandles;
             this.columnFamilyDescriptors = columnFamilyDescriptors;
             this.stateMetaInfoSnapshots = stateMetaInfoSnapshots;
-            this.readOptions = RocksDBOperationUtils.createTotalOrderSeekReadOptions();
+            this.readOptions = new ReadOptions();
         }
 
         @Override

flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBResource.java

@@ -165,7 +165,7 @@ protected void before() throws Throwable {
                         handlesToClose);
         this.writeOptions = new WriteOptions();
         this.writeOptions.disableWAL();
-        this.readOptions = RocksDBOperationUtils.createTotalOrderSeekReadOptions();
+        this.readOptions = new ReadOptions();
         this.columnFamilyHandles = new ArrayList<>(1);
         this.rocksDB =
                 RocksDB.open(

flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateMisuseOptionTest.java renamed to flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateOptionTest.java

@@ -52,27 +52,24 @@
 import static org.junit.Assert.assertTrue;
 
 /**
- * Tests to cover cases that even user misuse some options, RocksDB state-backend could still work
- * as expected or give explicit feedback.
+ * Tests to cover cases that if user choose options previously prone to misuse, embedded RocksDB
+ * state-backend could still work as expected or give explicit feedback.
  *
  * <p>RocksDB state-backend has some internal operations based on RocksDB's APIs which is
  * transparent for users. However, user could still configure options via {@link
  * RocksDBOptionsFactory}, and might lead some operations could not get expected result, e.g.
  * FLINK-17800
  */
-public class RocksDBStateMisuseOptionTest {
+public class RocksDBStateOptionTest {
 
     @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
 
     /**
-     * Tests to cover case when user misuse optimizeForPointLookup with iterator interfaces on map
+     * Tests to cover case when user choose optimizeForPointLookup with iterator interfaces on map
      * state.
-     *
-     * <p>The option {@link ColumnFamilyOptions#optimizeForPointLookup(long)} would lead to
-     * iterator.seek with prefix bytes invalid.
      */
     @Test
-    public void testMisuseOptimizePointLookupWithMapState() throws Exception {
+    public void testUseOptimizePointLookupWithMapState() throws Exception {
         RocksDBStateBackend rocksDBStateBackend = createStateBackendWithOptimizePointLookup();
         RocksDBKeyedStateBackend<Integer> keyedStateBackend =
                 createKeyedStateBackend(
@@ -111,14 +108,11 @@ public void testMisuseOptimizePointLookupWithMapState() throws Exception {
     }
 
     /**
-     * Tests to cover case when user misuse optimizeForPointLookup with peek operations on priority
+     * Tests to cover case when user choose optimizeForPointLookup with peek operations on priority
      * queue.
-     *
-     * <p>The option {@link ColumnFamilyOptions#optimizeForPointLookup(long)} would lead to
-     * iterator.seek with prefix bytes invalid.
      */
     @Test
-    public void testMisuseOptimizePointLookupWithPriorityQueue() throws IOException {
+    public void testUseOptimizePointLookupWithPriorityQueue() throws IOException {
         RocksDBStateBackend rocksDBStateBackend = createStateBackendWithOptimizePointLookup();
         RocksDBKeyedStateBackend<Integer> keyedStateBackend =
                 createKeyedStateBackend(