This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 953a5ff [hotfix][statebackend] Reduce and simplify code for column
creation in RocksDB backend
953a5ff is described below
commit 953a5ffcbdae4115f7d525f310723cf8770779df
Author: azagrebin <azagre...@users.noreply.github.com>
AuthorDate: Wed Mar 6 16:45:33 2019 +0100
[hotfix][statebackend] Reduce and simplify code for column creation in
RocksDB backend
This closes #7830.
---
.../streaming/state/RocksDBKeyedStateBackend.java | 16 +----
.../state/RocksDBKeyedStateBackendBuilder.java | 13 ++--
.../streaming/state/RocksDBOperationUtils.java | 82 ++++++++++++----------
.../state/RocksDBPriorityQueueSetFactory.java | 5 +-
.../restore/AbstractRocksDBRestoreOperation.java | 19 ++---
.../state/restore/RocksDBFullRestoreOperation.java | 9 +--
.../RocksDBIncrementalRestoreOperation.java | 36 +++-------
.../state/restore/RocksDBNoneRestoreOperation.java | 7 +-
.../state/ttl/RocksDbTtlCompactFiltersManager.java | 31 ++------
9 files changed, 78 insertions(+), 140 deletions(-)
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 c6d3863..4a6bd3c 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
@@ -126,9 +126,6 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
RocksDBKeyedStateBackend<K> backend) throws Exception;
}
- /** String that identifies the operator that owns this backend. */
- private final String operatorIdentifier;
-
/** Factory function to create column family options from state name. */
private final Function<String, ColumnFamilyOptions>
columnFamilyOptionsFactory;
@@ -158,9 +155,6 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
/** Number of bytes required to prefix the key groups. */
private final int keyGroupPrefixBytes;
- /** Thread number used to transfer state files while
restoring/snapshotting. */
- private final int numberOfTransferingThreads;
-
/**
* We are not using the default column family for Flink state ops, but
we still need to remember this handle so that
* we can close it properly when the backend is closed. Note that the
one returned by {@link RocksDB#open(String)}
@@ -201,7 +195,6 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
private final RocksDbTtlCompactFiltersManager ttlCompactFiltersManager;
public RocksDBKeyedStateBackend(
- String operatorIdentifier,
ClassLoader userCodeClassLoader,
File instanceBasePath,
DBOptions dbOptions,
@@ -211,7 +204,6 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
ExecutionConfig executionConfig,
- int numberOfTransferingThreads,
TtlTimeProvider ttlTimeProvider,
RocksDB db,
LinkedHashMap<String, RocksDbKvStateInfo> kvStateInformation,
@@ -233,10 +225,6 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
this.ttlCompactFiltersManager = ttlCompactFiltersManager;
- this.operatorIdentifier =
Preconditions.checkNotNull(operatorIdentifier);
-
- this.numberOfTransferingThreads = numberOfTransferingThreads;
-
// ensure that we use the right merge operator, because other
code relies on this
this.columnFamilyOptionsFactory =
Preconditions.checkNotNull(columnFamilyOptionsFactory);
@@ -500,8 +488,8 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
stateSerializer,
StateSnapshotTransformFactory.noTransform());
- newRocksStateInfo =
RocksDBOperationUtils.createStateInfo(newMetaInfo, ttlCompactFiltersManager,
- ttlTimeProvider, db,
columnFamilyOptionsFactory);
+ newRocksStateInfo =
RocksDBOperationUtils.createStateInfo(
+ newMetaInfo, db, columnFamilyOptionsFactory,
ttlCompactFiltersManager);
RocksDBOperationUtils.registerKvStateInformation(this.kvStateInformation,
this.nativeMetricMonitor,
stateDesc.getName(), newRocksStateInfo);
}
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
index 3f245d0..aa6845e 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
@@ -242,7 +242,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBacken
RocksDB db = null;
AbstractRocksDBRestoreOperation restoreOperation = null;
RocksDbTtlCompactFiltersManager ttlCompactFiltersManager =
- new
RocksDbTtlCompactFiltersManager(enableTtlCompactionFilter);
+ new
RocksDbTtlCompactFiltersManager(enableTtlCompactionFilter, ttlTimeProvider);
ResourceGuard rocksDBResourceGuard = new ResourceGuard();
SnapshotStrategy<K> snapshotStrategy;
@@ -324,7 +324,6 @@ public class RocksDBKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBacken
}
}
return new RocksDBKeyedStateBackend<>(
- this.operatorIdentifier,
this.userCodeClassLoader,
this.instanceBasePath,
this.dbOptions,
@@ -334,7 +333,6 @@ public class RocksDBKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBacken
this.numberOfKeyGroups,
this.keyGroupRange,
this.executionConfig,
- this.numberOfTransferingThreads,
this.ttlTimeProvider,
db,
kvStateInformation,
@@ -374,8 +372,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBacken
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
KeyedStateHandle firstStateHandle =
restoreStateHandles.iterator().next();
if (firstStateHandle instanceof IncrementalKeyedStateHandle) {
@@ -395,8 +392,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBacken
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
} else {
return new RocksDBFullRestoreOperation<>(
keyGroupRange,
@@ -413,8 +409,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends
AbstractKeyedStateBacken
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
}
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
index b183426..1e3fb8f 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
@@ -20,7 +20,6 @@ package org.apache.flink.contrib.streaming.state;
import org.apache.flink.configuration.ConfigConstants;
import
org.apache.flink.contrib.streaming.state.ttl.RocksDbTtlCompactFiltersManager;
import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -32,6 +31,8 @@ import org.rocksdb.DBOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -79,38 +80,11 @@ public class RocksDBOperationUtils {
return dbRef;
}
- public static ColumnFamilyDescriptor
createColumnFamilyDescriptor(String stateName, ColumnFamilyOptions
columnOptions) {
- byte[] nameBytes =
stateName.getBytes(ConfigConstants.DEFAULT_CHARSET);
-
Preconditions.checkState(!Arrays.equals(RocksDB.DEFAULT_COLUMN_FAMILY,
nameBytes),
- "The chosen state name 'default' collides with the name
of the default column family!");
-
- return new ColumnFamilyDescriptor(nameBytes, columnOptions);
- }
-
- public static ColumnFamilyHandle
createColumnFamily(ColumnFamilyDescriptor columnDescriptor, RocksDB db) {
- try {
- return db.createColumnFamily(columnDescriptor);
- } catch (RocksDBException e) {
- IOUtils.closeQuietly(columnDescriptor.getOptions());
- throw new FlinkRuntimeException("Error creating
ColumnFamilyHandle.", e);
- }
- }
-
- public static ColumnFamilyHandle createColumnFamily(
- String stateName,
- Function<String, ColumnFamilyOptions>
columnFamilyOptionsFactory,
- RocksDB db) {
- ColumnFamilyOptions options =
createColumnFamilyOptions(columnFamilyOptionsFactory, stateName);
- return
createColumnFamily(createColumnFamilyDescriptor(stateName, options), db);
- }
-
public static RocksIteratorWrapper getRocksIterator(RocksDB db) {
return new RocksIteratorWrapper(db.newIterator());
}
- public static RocksIteratorWrapper getRocksIterator(
- RocksDB db,
- ColumnFamilyHandle columnFamilyHandle) {
+ public static RocksIteratorWrapper getRocksIterator(RocksDB db,
ColumnFamilyHandle columnFamilyHandle) {
return new
RocksIteratorWrapper(db.newIterator(columnFamilyHandle));
}
@@ -119,8 +93,8 @@ public class RocksDBOperationUtils {
RocksDBNativeMetricMonitor nativeMetricMonitor,
String columnFamilyName,
RocksDBKeyedStateBackend.RocksDbKvStateInfo registeredColumn) {
- kvStateInformation.put(columnFamilyName, registeredColumn);
+ kvStateInformation.put(columnFamilyName, registeredColumn);
if (nativeMetricMonitor != null) {
nativeMetricMonitor.registerColumnFamily(columnFamilyName,
registeredColumn.columnFamilyHandle);
}
@@ -128,26 +102,58 @@ public class RocksDBOperationUtils {
/**
* Creates a state info from a new meta info to use with a k/v state.
+ *
+ * <p>Creates the column family for the state.
+ * Sets TTL compaction filter if {@code ttlCompactFiltersManager} is
not {@code null}.
*/
public static RocksDBKeyedStateBackend.RocksDbKvStateInfo
createStateInfo(
RegisteredStateMetaInfoBase metaInfoBase,
- RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider,
RocksDB db,
- Function<String, ColumnFamilyOptions>
columnFamilyOptionsFactory) {
- ColumnFamilyOptions options =
createColumnFamilyOptions(columnFamilyOptionsFactory, metaInfoBase.getName());
-
ttlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(ttlTimeProvider,
metaInfoBase, options);
- ColumnFamilyDescriptor columnFamilyDescriptor =
createColumnFamilyDescriptor(metaInfoBase.getName(), options);
+ Function<String, ColumnFamilyOptions>
columnFamilyOptionsFactory,
+ @Nullable RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager) {
+
+ ColumnFamilyDescriptor columnFamilyDescriptor =
createColumnFamilyDescriptor(
+ metaInfoBase, columnFamilyOptionsFactory,
ttlCompactFiltersManager);
return new
RocksDBKeyedStateBackend.RocksDbKvStateInfo(createColumnFamily(columnFamilyDescriptor,
db), metaInfoBase);
}
- public static ColumnFamilyOptions createColumnFamilyOptions(
+ /**
+ * Creates a column descriptor for sate column family.
+ *
+ * <p>Sets TTL compaction filter if {@code ttlCompactFiltersManager} is
not {@code null}.
+ */
+ public static ColumnFamilyDescriptor createColumnFamilyDescriptor(
+ RegisteredStateMetaInfoBase metaInfoBase,
Function<String, ColumnFamilyOptions>
columnFamilyOptionsFactory,
- String stateName) {
+ @Nullable RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager) {
+
+ ColumnFamilyOptions options =
createColumnFamilyOptions(columnFamilyOptionsFactory, metaInfoBase.getName());
+ if (ttlCompactFiltersManager != null) {
+
ttlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(metaInfoBase,
options);
+ }
+ byte[] nameBytes =
metaInfoBase.getName().getBytes(ConfigConstants.DEFAULT_CHARSET);
+
Preconditions.checkState(!Arrays.equals(RocksDB.DEFAULT_COLUMN_FAMILY,
nameBytes),
+ "The chosen state name 'default' collides with the name
of the default column family!");
+
+ return new ColumnFamilyDescriptor(nameBytes, options);
+ }
+
+ public static ColumnFamilyOptions createColumnFamilyOptions(
+ Function<String, ColumnFamilyOptions>
columnFamilyOptionsFactory, String stateName) {
+
// ensure that we use the right merge operator, because other
code relies on this
return
columnFamilyOptionsFactory.apply(stateName).setMergeOperatorName(MERGE_OPERATOR_NAME);
}
+ private static ColumnFamilyHandle
createColumnFamily(ColumnFamilyDescriptor columnDescriptor, RocksDB db) {
+ try {
+ return db.createColumnFamily(columnDescriptor);
+ } catch (RocksDBException e) {
+ IOUtils.closeQuietly(columnDescriptor.getOptions());
+ throw new FlinkRuntimeException("Error creating
ColumnFamilyHandle.", e);
+ }
+ }
+
public static void addColumnFamilyOptionsToCloseLater(
List<ColumnFamilyOptions> columnFamilyOptions,
ColumnFamilyHandle columnFamilyHandle) {
try {
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
index 1eb0f47..1cbeebe 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
@@ -145,12 +145,9 @@ public class RocksDBPriorityQueueSetFactory implements
PriorityQueueSetFactory {
if (stateInfo == null) {
// Currently this class is for timer service and TTL
feature is not applicable here,
// so no need to register compact filter when creating
column family
- final ColumnFamilyHandle columnFamilyHandle =
-
RocksDBOperationUtils.createColumnFamily(stateName, columnFamilyOptionsFactory,
this.db);
RegisteredPriorityQueueStateBackendMetaInfo<T> metaInfo
=
new
RegisteredPriorityQueueStateBackendMetaInfo<>(stateName,
byteOrderedElementSerializer);
-
- stateInfo = new
RocksDBKeyedStateBackend.RocksDbKvStateInfo(columnFamilyHandle, metaInfo);
+ stateInfo =
RocksDBOperationUtils.createStateInfo(metaInfo, db, columnFamilyOptionsFactory,
null);
RocksDBOperationUtils.registerKvStateInformation(kvStateInformation,
nativeMetricMonitor, stateName, stateInfo);
} else {
// TODO we implement the simple way of supporting the
current functionality, mimicking keyed state
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
index 9a29b62..064483c 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
@@ -33,7 +33,6 @@ import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.StateMigrationException;
@@ -42,7 +41,6 @@ import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
import javax.annotation.Nonnull;
@@ -87,7 +85,6 @@ public abstract class AbstractRocksDBRestoreOperation<K>
implements RocksDBResto
// - Full restore
// - data ingestion after db open:
#getOrRegisterStateColumnFamilyHandle before creating column family
protected final RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager;
- protected final TtlTimeProvider ttlTimeProvider;
protected RocksDB db;
protected ColumnFamilyHandle defaultColumnFamilyHandle;
@@ -109,8 +106,7 @@ public abstract class AbstractRocksDBRestoreOperation<K>
implements RocksDBResto
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> stateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider) {
+ @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager) {
this.keyGroupRange = keyGroupRange;
this.keyGroupPrefixBytes = keyGroupPrefixBytes;
this.numberOfTransferringThreads = numberOfTransferringThreads;
@@ -127,12 +123,11 @@ public abstract class AbstractRocksDBRestoreOperation<K>
implements RocksDBResto
this.metricGroup = metricGroup;
this.restoreStateHandles = stateHandles;
this.ttlCompactFiltersManager = ttlCompactFiltersManager;
- this.ttlTimeProvider = ttlTimeProvider;
this.columnFamilyHandles = new ArrayList<>(1);
this.columnFamilyDescriptors = Collections.emptyList();
}
- public void openDB() throws IOException {
+ void openDB() throws IOException {
db = RocksDBOperationUtils.openDB(
dbPath,
columnFamilyDescriptors,
@@ -150,9 +145,9 @@ public abstract class AbstractRocksDBRestoreOperation<K>
implements RocksDBResto
return this.db;
}
- protected RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
+ RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
ColumnFamilyHandle columnFamilyHandle,
- StateMetaInfoSnapshot stateMetaInfoSnapshot) throws
RocksDBException {
+ StateMetaInfoSnapshot stateMetaInfoSnapshot) {
RocksDbKvStateInfo registeredStateMetaInfoEntry =
kvStateInformation.get(stateMetaInfoSnapshot.getName());
@@ -163,8 +158,8 @@ public abstract class AbstractRocksDBRestoreOperation<K>
implements RocksDBResto
RegisteredStateMetaInfoBase stateMetaInfo =
RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
if (columnFamilyHandle == null) {
- registeredStateMetaInfoEntry =
RocksDBOperationUtils.createStateInfo(stateMetaInfo,
- ttlCompactFiltersManager,
ttlTimeProvider, db, columnFamilyOptionsFactory);
+ registeredStateMetaInfoEntry =
RocksDBOperationUtils.createStateInfo(
+ stateMetaInfo, db,
columnFamilyOptionsFactory, ttlCompactFiltersManager);
} else {
registeredStateMetaInfoEntry = new
RocksDbKvStateInfo(columnFamilyHandle, stateMetaInfo);
}
@@ -181,7 +176,7 @@ public abstract class AbstractRocksDBRestoreOperation<K>
implements RocksDBResto
return registeredStateMetaInfoEntry;
}
- protected KeyedBackendSerializationProxy<K> readMetaData(DataInputView
dataInputView)
+ KeyedBackendSerializationProxy<K> readMetaData(DataInputView
dataInputView)
throws IOException, StateMigrationException {
// isSerializerPresenceRequired flag is set to false, since for
the RocksDB state backend,
// deserialization of state happens lazily during runtime; we
depend on the fact
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
index e9c8fe9..289ebdc 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
@@ -38,7 +38,6 @@ import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.StreamCompressionDecorator;
import org.apache.flink.runtime.state.UncompressedStreamCompressionDecorator;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StateMigrationException;
@@ -103,8 +102,7 @@ public class RocksDBFullRestoreOperation<K> extends
AbstractRocksDBRestoreOperat
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> restoreStateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider) {
+ @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager) {
super(
keyGroupRange,
keyGroupPrefixBytes,
@@ -120,8 +118,7 @@ public class RocksDBFullRestoreOperation<K> extends
AbstractRocksDBRestoreOperat
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
/**
@@ -169,7 +166,7 @@ public class RocksDBFullRestoreOperation<K> extends
AbstractRocksDBRestoreOperat
/**
* Restore the KV-state / ColumnFamily meta data for all key-groups
referenced by the current state handle.
*/
- private void restoreKVStateMetaData() throws IOException,
StateMigrationException, RocksDBException {
+ private void restoreKVStateMetaData() throws IOException,
StateMigrationException {
KeyedBackendSerializationProxy<K> serializationProxy =
readMetaData(currentStateHandleInView);
this.keygroupStreamCompressionDecorator =
serializationProxy.isUsingKeyGroupCompression() ?
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index 3371f4f..9088006 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -18,7 +18,6 @@
package org.apache.flink.contrib.streaming.state.restore;
-import org.apache.flink.configuration.ConfigConstants;
import
org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils;
import org.apache.flink.contrib.streaming.state.RocksDBKeySerializationUtils;
import
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.RocksDbKvStateInfo;
@@ -44,11 +43,11 @@ import
org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedBackendSerializationProxy;
import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
import org.apache.flink.runtime.state.StateHandleID;
import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.IOUtils;
import org.rocksdb.ColumnFamilyDescriptor;
@@ -106,8 +105,7 @@ public class RocksDBIncrementalRestoreOperation<K> extends
AbstractRocksDBRestor
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> restoreStateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider) {
+ @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager) {
super(keyGroupRange,
keyGroupPrefixBytes,
numberOfTransferringThreads,
@@ -122,8 +120,7 @@ public class RocksDBIncrementalRestoreOperation<K> extends
AbstractRocksDBRestor
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
this.operatorIdentifier = operatorIdentifier;
this.restoredSstFiles = new TreeMap<>();
this.lastCompletedCheckpointId = -1L;
@@ -252,19 +249,10 @@ public class RocksDBIncrementalRestoreOperation<K>
extends AbstractRocksDBRestor
}
}
- private void registerColumnFamilyHandles(List<StateMetaInfoSnapshot>
metaInfoSnapshots)
- throws BackendBuildingException {
+ private void registerColumnFamilyHandles(List<StateMetaInfoSnapshot>
metaInfoSnapshots) {
// Register CF handlers
for (int i = 0; i < metaInfoSnapshots.size(); ++i) {
- try {
- getOrRegisterStateColumnFamilyHandle(
- columnFamilyHandles.get(i),
- metaInfoSnapshots.get(i));
- } catch (RocksDBException e) {
- String errMsg = "Failed to register CF handle.";
- LOG.error(errMsg, e);
- throw new BackendBuildingException(errMsg, e);
- }
+
getOrRegisterStateColumnFamilyHandle(columnFamilyHandles.get(i),
metaInfoSnapshots.get(i));
}
}
@@ -452,16 +440,10 @@ public class RocksDBIncrementalRestoreOperation<K>
extends AbstractRocksDBRestor
new ArrayList<>(stateMetaInfoSnapshots.size());
for (StateMetaInfoSnapshot stateMetaInfoSnapshot :
stateMetaInfoSnapshots) {
- ColumnFamilyOptions options =
RocksDBOperationUtils.createColumnFamilyOptions(
- columnFamilyOptionsFactory,
stateMetaInfoSnapshot.getName());
- if (registerTtlCompactFilter) {
-
ttlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(ttlTimeProvider,
- stateMetaInfoSnapshot, options);
- }
- ColumnFamilyDescriptor columnFamilyDescriptor = new
ColumnFamilyDescriptor(
-
stateMetaInfoSnapshot.getName().getBytes(ConfigConstants.DEFAULT_CHARSET),
- options);
-
+ RegisteredStateMetaInfoBase metaInfoBase =
+
RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
+ ColumnFamilyDescriptor columnFamilyDescriptor =
RocksDBOperationUtils.createColumnFamilyDescriptor(
+ metaInfoBase, columnFamilyOptionsFactory,
registerTtlCompactFilter ? ttlCompactFiltersManager : null);
columnFamilyDescriptors.add(columnFamilyDescriptor);
}
return columnFamilyDescriptors;
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
index 873132a..bfc25d6 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
@@ -26,7 +26,6 @@ import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.StateSerializerProvider;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
@@ -57,8 +56,7 @@ public class RocksDBNoneRestoreOperation<K> extends
AbstractRocksDBRestoreOperat
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> restoreStateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider
+ @Nonnull RocksDbTtlCompactFiltersManager
ttlCompactFiltersManager
) {
super(keyGroupRange,
keyGroupPrefixBytes,
@@ -74,8 +72,7 @@ public class RocksDBNoneRestoreOperation<K> extends
AbstractRocksDBRestoreOperat
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
@Override
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
index ccdb092..affea5b 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
@@ -23,12 +23,10 @@ import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.ListSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo;
import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.runtime.state.ttl.TtlStateFactory;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.runtime.state.ttl.TtlUtils;
@@ -57,47 +55,30 @@ public class RocksDbTtlCompactFiltersManager {
/** Enables RocksDb compaction filter for State with TTL. */
private final boolean enableTtlCompactionFilter;
+ private final TtlTimeProvider ttlTimeProvider;
+
/** Registered compaction filter factories. */
private final LinkedHashMap<String, FlinkCompactionFilterFactory>
compactionFilterFactories;
- public RocksDbTtlCompactFiltersManager(boolean
enableTtlCompactionFilter) {
+ public RocksDbTtlCompactFiltersManager(boolean
enableTtlCompactionFilter, TtlTimeProvider ttlTimeProvider) {
this.enableTtlCompactionFilter = enableTtlCompactionFilter;
+ this.ttlTimeProvider = ttlTimeProvider;
this.compactionFilterFactories = new LinkedHashMap<>();
}
public void setAndRegisterCompactFilterIfStateTtl(
- TtlTimeProvider ttlTimeProvider,
- @Nonnull StateMetaInfoSnapshot stateMetaInfoSnapshot,
- @Nonnull ColumnFamilyOptions options) {
-
- boolean keyValueState =
stateMetaInfoSnapshot.getBackendStateType() ==
StateMetaInfoSnapshot.BackendStateType.KEY_VALUE;
- if (enableTtlCompactionFilter && keyValueState) {
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<?> stateSerializerSnapshot =
Preconditions.checkNotNull(
- (TypeSerializerSnapshot<?>)
stateMetaInfoSnapshot.getTypeSerializerSnapshot(
-
StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
- TypeSerializer<?> serializer =
stateSerializerSnapshot.restoreSerializer();
- if
(TtlStateFactory.TtlSerializer.isTtlStateSerializer(serializer)) {
-
createAndSetCompactFilterFactory(stateMetaInfoSnapshot.getName(),
ttlTimeProvider, options);
- }
- }
- }
-
- public void setAndRegisterCompactFilterIfStateTtl(
- TtlTimeProvider ttlTimeProvider,
@Nonnull RegisteredStateMetaInfoBase metaInfoBase,
@Nonnull ColumnFamilyOptions options) {
if (enableTtlCompactionFilter && metaInfoBase instanceof
RegisteredKeyValueStateBackendMetaInfo) {
RegisteredKeyValueStateBackendMetaInfo kvMetaInfoBase =
(RegisteredKeyValueStateBackendMetaInfo) metaInfoBase;
if
(TtlStateFactory.TtlSerializer.isTtlStateSerializer(kvMetaInfoBase.getStateSerializer()))
{
-
createAndSetCompactFilterFactory(metaInfoBase.getName(), ttlTimeProvider,
options);
+
createAndSetCompactFilterFactory(metaInfoBase.getName(), options);
}
}
}
- private void createAndSetCompactFilterFactory(
- String stateName, TtlTimeProvider ttlTimeProvider, @Nonnull
ColumnFamilyOptions options) {
+ private void createAndSetCompactFilterFactory(String stateName,
@Nonnull ColumnFamilyOptions options) {
FlinkCompactionFilterFactory compactionFilterFactory =
new FlinkCompactionFilterFactory(new
TimeProviderWrapper(ttlTimeProvider), createRocksDbNativeLogger());
