[FLINK-14482][rocksdb] Bump FRocksDB version to 6.20.3

This closes apache#16794.

Author: autophagy

docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html

@@ -48,7 +48,7 @@
             <td><h5>state.backend.rocksdb.compaction.style</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td><p>Enum</p></td>
-            <td>The specified compaction style for DB. Candidate compaction style is LEVEL, FIFO or UNIVERSAL, and RocksDB choose 'LEVEL' as default style.<br /><br />Possible values:<ul><li>"LEVEL"</li><li>"UNIVERSAL"</li><li>"FIFO"</li></ul></td>
+            <td>The specified compaction style for DB. Candidate compaction style is LEVEL, FIFO, UNIVERSAL or NONE, and RocksDB choose 'LEVEL' as default style.<br /><br />Possible values:<ul><li>"LEVEL"</li><li>"UNIVERSAL"</li><li>"FIFO"</li><li>"NONE"</li></ul></td>
         </tr>
         <tr>
             <td><h5>state.backend.rocksdb.files.open</h5></td>

flink-dist/src/main/resources/META-INF/NOTICE

@@ -6,7 +6,7 @@ The Apache Software Foundation (http://www.apache.org/).
 
 This project bundles the following dependencies under the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
 
-- com.data-artisans:frocksdbjni:5.17.2-artisans-2.0
+- com.ververica:frocksdbjni:6.20.3-ververica-1.0
 - com.google.code.findbugs:jsr305:1.3.9
 - com.twitter:chill-java:0.7.6
 - com.twitter:chill_2.11:0.7.6

flink-state-backends/flink-statebackend-rocksdb/pom.xml

@@ -57,7 +57,7 @@ under the License.
 		<dependency>
 			<groupId>com.ververica</groupId>
 			<artifactId>frocksdbjni</artifactId>
-			<version>5.17.2-ververica-2.1</version>
+			<version>6.20.3-ververica-1.0</version>
 		</dependency>
 
 		<!-- test dependencies -->

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

@@ -31,6 +31,7 @@
 import static org.apache.flink.configuration.description.LinkElement.link;
 import static org.rocksdb.CompactionStyle.FIFO;
 import static org.rocksdb.CompactionStyle.LEVEL;
+import static org.rocksdb.CompactionStyle.NONE;
 import static org.rocksdb.CompactionStyle.UNIVERSAL;
 import static org.rocksdb.InfoLogLevel.DEBUG_LEVEL;
 import static org.rocksdb.InfoLogLevel.ERROR_LEVEL;
@@ -102,9 +103,13 @@ public class RocksDBConfigurableOptions implements Serializable {
                     .noDefaultValue()
                     .withDescription(
                             String.format(
-                                    "The specified compaction style for DB. Candidate compaction style is %s, %s or %s, "
+                                    "The specified compaction style for DB. Candidate compaction style is %s, %s, %s or %s, "
                                             + "and RocksDB choose '%s' as default style.",
-                                    LEVEL.name(), FIFO.name(), UNIVERSAL.name(), LEVEL.name()));
+                                    LEVEL.name(),
+                                    FIFO.name(),
+                                    UNIVERSAL.name(),
+                                    NONE.name(),
+                                    LEVEL.name()));
 
     public static final ConfigOption<Boolean> USE_DYNAMIC_LEVEL_SIZE =
             key("state.backend.rocksdb.compaction.level.use-dynamic-size")

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

@@ -250,7 +250,7 @@ private boolean overwriteFilterIfExist(BlockBasedTableConfig blockBasedTableConf
     @VisibleForTesting
     static Filter getFilterFromBlockBasedTableConfig(BlockBasedTableConfig blockBasedTableConfig)
             throws NoSuchFieldException, IllegalAccessException {
-        Field filterField = blockBasedTableConfig.getClass().getDeclaredField("filter_");
+        Field filterField = blockBasedTableConfig.getClass().getDeclaredField("filterPolicy");
         filterField.setAccessible(true);
         Object filter = filterField.get(blockBasedTableConfig);
         filterField.setAccessible(false);

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

@@ -27,6 +27,7 @@
 import javax.annotation.Nonnull;
 
 import java.io.Closeable;
+import java.nio.ByteBuffer;
 
 /**
  * This is a wrapper around {@link RocksIterator} to check the iterator status for all the methods
@@ -74,6 +75,18 @@ public void seekForPrev(byte[] target) {
         status();
     }
 
+    @Override
+    public void seek(ByteBuffer target) {
+        iterator.seek(target);
+        status();
+    }
+
+    @Override
+    public void seekForPrev(ByteBuffer target) {
+        iterator.seekForPrev(target);
+        status();
+    }
+
     @Override
     public void next() {
         iterator.next();
@@ -95,6 +108,12 @@ public void status() {
         }
     }
 
+    @Override
+    public void refresh() throws RocksDBException {
+        iterator.refresh();
+        status();
+    }
+
     public byte[] key() {
         return iterator.key();
     }

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

@@ -28,7 +28,7 @@
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
-import org.rocksdb.RocksDB;
+import org.rocksdb.NativeLibraryLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,7 +39,7 @@
 
 /** Tests for {@link EmbeddedRocksDBStateBackend} on initialization. */
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({RocksDB.class})
+@PrepareForTest({NativeLibraryLoader.class})
 public class RocksDBInitTest {
 
     private static final Logger LOG = LoggerFactory.getLogger(RocksDBInitTest.class);
@@ -57,8 +57,9 @@ public void testResetInitFlag() throws Exception {
 
     @Test
     public void testTempLibFolderDeletedOnFail() throws Exception {
-        PowerMockito.spy(RocksDB.class);
-        PowerMockito.when(RocksDB.class, "loadLibrary").thenThrow(new ExpectedTestException());
+        PowerMockito.spy(NativeLibraryLoader.class);
+        PowerMockito.when(NativeLibraryLoader.class, "getInstance")
+                .thenThrow(new ExpectedTestException());
 
         File tempFolder = temporaryFolder.newFolder();
         try {

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

@@ -173,15 +173,15 @@ private Cache getBlockCache(ColumnFamilyOptions columnOptions) {
         }
         Field cacheField = null;
         try {
-            cacheField = BlockBasedTableConfig.class.getDeclaredField("blockCache_");
+            cacheField = BlockBasedTableConfig.class.getDeclaredField("blockCache");
         } catch (NoSuchFieldException e) {
-            fail("blockCache_ is not defined");
+            fail("blockCache is not defined");
         }
         cacheField.setAccessible(true);
         try {
             return (Cache) cacheField.get(blockBasedTableConfig);
         } catch (IllegalAccessException e) {
-            fail("Cannot access blockCache_ field.");
+            fail("Cannot access blockCache field.");
             return null;
         }
     }