[FLINK-15286][state-backend-rocksdb] Read option whether to use manag…

…ed memory from configuration This closes apache#10600
spotify · Dec 17, 2019 · 685c443 · 685c443
1 parent b0783b5
commit 685c443
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 0 deletions.
diff --git a/...db/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMemoryConfiguration.java b/...db/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMemoryConfiguration.java
@@ -181,6 +181,10 @@ public static RocksDBMemoryConfiguration fromOtherAndConfiguration(
 
  final RocksDBMemoryConfiguration newConfig = new RocksDBMemoryConfiguration();
 
+ newConfig.useManagedMemory = other.useManagedMemory != null
+ ? other.useManagedMemory
+ : config.getBoolean(RocksDBOptions.USE_MANAGED_MEMORY);
+
  newConfig.fixedMemoryPerSlot = other.fixedMemoryPerSlot != null
  ? other.fixedMemoryPerSlot
  : config.get(RocksDBOptions.FIX_PER_SLOT_MEMORY_SIZE);

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
@@ -625,6 +625,7 @@ public void testRocksDbReconfigurationCopiesExistingValues() throws Exception {
  @Test
  public void testDefaultMemoryControlParameters() {
  RocksDBMemoryConfiguration memSettings = new RocksDBMemoryConfiguration();
+ assertFalse(memSettings.isUsingManagedMemory());
  assertFalse(memSettings.isUsingFixedMemoryPerSlot());
  assertEquals(RocksDBOptions.HIGH_PRIORITY_POOL_RATIO.defaultValue(), memSettings.getHighPriorityPoolRatio(), 0.0);
  assertEquals(RocksDBOptions.WRITE_BUFFER_RATIO.defaultValue(), memSettings.getWriteBufferRatio(), 0.0);
@@ -635,6 +636,17 @@ public void testDefaultMemoryControlParameters() {
  assertEquals(RocksDBOptions.WRITE_BUFFER_RATIO.defaultValue(), configured.getWriteBufferRatio(), 0.0);
  }
 
+ @Test
+ public void testConfigureManagedMemory() {
+ final Configuration config = new Configuration();
+ config.setBoolean(RocksDBOptions.USE_MANAGED_MEMORY, true);
+
+ final RocksDBMemoryConfiguration memSettings = RocksDBMemoryConfiguration.fromOtherAndConfiguration(
+ new RocksDBMemoryConfiguration(), config);
+
+ assertTrue(memSettings.isUsingManagedMemory());
+ }
+
  @Test
  public void testConfigureIllegalMemoryControlParameters() {
  RocksDBMemoryConfiguration memSettings = new RocksDBMemoryConfiguration();