You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/04/06 10:32:22 UTC
[3/3] flink git commit: [FLINK-8968][state] Pull the creation of
readOptions out of loop to avoid native resource leak.
[FLINK-8968][state] Pull the creation of readOptions out of loop to avoid native resource leak.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f5071d7a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f5071d7a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f5071d7a
Branch: refs/heads/master
Commit: f5071d7a6128f78e7f0ab7ccd8813d30543fb3a2
Parents: 2d87244
Author: sihuazhou <su...@163.com>
Authored: Thu Mar 15 23:57:11 2018 +0800
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Fri Apr 6 12:32:08 2018 +0200
----------------------------------------------------------------------
.../contrib/streaming/state/RocksDBKeyedStateBackend.java | 9 +++++----
.../streaming/state/RocksDBStateBackendConfigTest.java | 8 ++++++--
2 files changed, 11 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f5071d7a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 6a23181..cdeb608 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -1952,15 +1952,16 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
this.kvStateIterators = new ArrayList<>(kvStateInformationCopy.size());
int kvStateId = 0;
+
+ //retrieve iterator for this k/v states
+ readOptions = new ReadOptions();
+ readOptions.setSnapshot(snapshot);
+
for (Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>> column :
kvStateInformationCopy) {
metaInfoSnapshots.add(column.f1.snapshot());
- //retrieve iterator for this k/v states
- readOptions = new ReadOptions();
- readOptions.setSnapshot(snapshot);
-
kvStateIterators.add(
new Tuple2<>(stateBackend.db.newIterator(column.f0, readOptions), kvStateId));
http://git-wip-us.apache.org/repos/asf/flink/blob/f5071d7a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
----------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
index 2dd67f5..65d5b2e 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
@@ -299,7 +299,9 @@ public class RocksDBStateBackendConfigTest {
});
assertNotNull(rocksDbBackend.getOptions());
- assertEquals(CompactionStyle.FIFO, rocksDbBackend.getColumnOptions().compactionStyle());
+ try (ColumnFamilyOptions colCreated = rocksDbBackend.getColumnOptions()) {
+ assertEquals(CompactionStyle.FIFO, colCreated.compactionStyle());
+ }
}
@Test
@@ -324,7 +326,9 @@ public class RocksDBStateBackendConfigTest {
assertEquals(PredefinedOptions.SPINNING_DISK_OPTIMIZED, rocksDbBackend.getPredefinedOptions());
assertNotNull(rocksDbBackend.getOptions());
- assertEquals(CompactionStyle.UNIVERSAL, rocksDbBackend.getColumnOptions().compactionStyle());
+ try (ColumnFamilyOptions colCreated = rocksDbBackend.getColumnOptions()) {
+ assertEquals(CompactionStyle.UNIVERSAL, colCreated.compactionStyle());
+ }
}
@Test