aljoscha commented on a change in pull request #14913:
URL: https://github.com/apache/flink/pull/14913#discussion_r573919950
##########
File path:
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
##########
@@ -94,53 +81,44 @@
// - Full restore
// - data ingestion after db open: #getOrRegisterStateColumnFamilyHandle
before creating
// column family
- protected final RocksDbTtlCompactFiltersManager ttlCompactFiltersManager;
-
- protected RocksDB db;
- protected ColumnFamilyHandle defaultColumnFamilyHandle;
- protected RocksDBNativeMetricMonitor nativeMetricMonitor;
- protected boolean isKeySerializerCompatibilityChecked;
- protected final Long writeBufferManagerCapacity;
-
- protected AbstractRocksDBRestoreOperation(
- KeyGroupRange keyGroupRange,
- int keyGroupPrefixBytes,
- int numberOfTransferringThreads,
- CloseableRegistry cancelStreamRegistry,
- ClassLoader userCodeClassLoader,
- Map<String, RocksDbKvStateInfo> kvStateInformation,
- StateSerializerProvider<K> keySerializerProvider,
- File instanceBasePath,
+ private final RocksDbTtlCompactFiltersManager ttlCompactFiltersManager;
+
+ private RocksDB db;
+ private ColumnFamilyHandle defaultColumnFamilyHandle;
+ private RocksDBNativeMetricMonitor nativeMetricMonitor;
+ private final Long writeBufferManagerCapacity;
+
+ protected RocksDBHandle(
+ Map<String, RocksDBKeyedStateBackend.RocksDbKvStateInfo>
kvStateInformation,
File instanceRocksDBPath,
DBOptions dbOptions,
Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
- @Nonnull Collection<KeyedStateHandle> stateHandles,
@Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
Long writeBufferManagerCapacity) {
- this.keyGroupRange = keyGroupRange;
- this.keyGroupPrefixBytes = keyGroupPrefixBytes;
- this.numberOfTransferringThreads = numberOfTransferringThreads;
- this.cancelStreamRegistry = cancelStreamRegistry;
- this.userCodeClassLoader = userCodeClassLoader;
this.kvStateInformation = kvStateInformation;
- this.keySerializerProvider = keySerializerProvider;
- this.instanceBasePath = instanceBasePath;
- this.instanceRocksDBPath = instanceRocksDBPath;
this.dbPath = instanceRocksDBPath.getAbsolutePath();
this.dbOptions = dbOptions;
this.columnFamilyOptionsFactory = columnFamilyOptionsFactory;
this.nativeMetricOptions = nativeMetricOptions;
this.metricGroup = metricGroup;
- this.restoreStateHandles = stateHandles;
this.ttlCompactFiltersManager = ttlCompactFiltersManager;
this.columnFamilyHandles = new ArrayList<>(1);
this.columnFamilyDescriptors = Collections.emptyList();
this.writeBufferManagerCapacity = writeBufferManagerCapacity;
}
void openDB() throws IOException {
+ openDB(Collections.emptyList(), Collections.emptyList());
+ }
+
+ void openDB(
Review comment:
This is only ever called together with
`restoreInstanceDirectoryFromPath()`. Should one of them become an internal
method?
##########
File path:
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
##########
@@ -49,39 +41,34 @@
import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
+import static
org.apache.flink.contrib.streaming.state.snapshot.RocksSnapshotUtil.SST_FILE_SUFFIX;
+
/**
- * Base implementation of RocksDB restore operation.
- *
- * @param <K> The data type that the serializer serializes.
+ * Utility for creating a RocksDB instance from a restored state and
populating {@link
+ *
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.RocksDbKvStateInfo}.
*/
-public abstract class AbstractRocksDBRestoreOperation<K>
- implements RocksDBRestoreOperation, AutoCloseable {
+class RocksDBHandle implements AutoCloseable {
Review comment:
I'm not sure I like the name "Handle" but I don't have a better option
for now. 😅
##########
File path:
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java
##########
@@ -49,39 +41,34 @@
import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
+import static
org.apache.flink.contrib.streaming.state.snapshot.RocksSnapshotUtil.SST_FILE_SUFFIX;
+
/**
- * Base implementation of RocksDB restore operation.
- *
- * @param <K> The data type that the serializer serializes.
+ * Utility for creating a RocksDB instance from a restored state and
populating {@link
+ *
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.RocksDbKvStateInfo}.
Review comment:
```suggestion
* Utility for creating a RocksDB instance either from scratch or from
restored local state. This will also register {@link RocksDbKvStateInfo} when
using {@link #openDB(...)}.
```
Sorry, not completely correct Javadocs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
