dawidwys commented on a change in pull request #14982:
URL: https://github.com/apache/flink/pull/14982#discussion_r581849390
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java
##########
@@ -245,6 +262,24 @@ void snapshotState(
}
}
+ @Nonnull
+ public static SnapshotStrategyRunner<KeyedStateHandle, ? extends
FullSnapshotResources<?>>
+ prepareSavepoint(
+ CheckpointableKeyedStateBackend<?> keyedStateBackend,
+ CloseableRegistry closeableRegistry)
+ throws Exception {
+ FullSnapshotResources<?> savepointResources =
keyedStateBackend.savepoint();
+
+ SavepointSnapshotStrategy<?> savepointSnapshotStrategy =
+ new SavepointSnapshotStrategy<>(savepointResources);
+
+ return new SnapshotStrategyRunner<>(
+ "Asynchronous full Savepoint",
+ savepointSnapshotStrategy,
+ closeableRegistry,
+ ASYNCHRONOUS);
Review comment:
Do we want to hardcode `ASYNCHRONOUS` here?
##########
File path:
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
##########
@@ -500,9 +487,18 @@ private static void checkAndCreateDirectory(File
directory) throws IOException {
lastCompletedCheckpointId,
numberOfTransferingThreads);
} else {
- checkpointSnapshotStrategy = savepointSnapshotStrategy;
+ checkpointSnapshotStrategy =
+ new RocksFullSnapshotStrategy<>(
+ db,
+ rocksDBResourceGuard,
+ keySerializerProvider.currentSchemaSerializer(),
+ kvStateInformation,
+ keyGroupRange,
+ keyGroupPrefixBytes,
+ localRecoveryConfig,
+ keyGroupCompressionDecorator);
}
- return new SnapshotStrategy<>(checkpointSnapshotStrategy,
savepointSnapshotStrategy);
+ return new SnapshotStrategy<>(checkpointSnapshotStrategy);
Review comment:
We can get rid of the `SnapshotStrategy` wrapper.
##########
File path:
flink-tests/src/test/java/org/apache/flink/test/state/HeapSavepointStateBackendSwitchTest.java
##########
@@ -33,7 +33,7 @@ public HeapSavepointStateBackendSwitchTest(BackendSwitchSpec
toBackend) {
super(BackendSwitchSpecs.HEAP, toBackend);
}
- @Parameterized.Parameters
+ @Parameterized.Parameters(name = "Target Backend: {0}")
Review comment:
Good improvement!
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java
##########
@@ -174,6 +180,7 @@ public OperatorSnapshotFutures snapshotState(
return snapshotInProgress;
}
+ @SuppressWarnings("checkstyle:RegexpSingleline")
Review comment:
What is this suppression for?
----------------------------------------------------------------
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:
us...@infra.apache.org
