Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/3801#discussion_r114325070
--- Diff:
flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
---
@@ -265,9 +281,64 @@ private boolean hasRegisteredState() {
final CheckpointStreamFactory streamFactory,
CheckpointOptions checkpointOptions) throws Exception {
+ if (checkpointOptions.getCheckpointType() ==
CheckpointOptions.CheckpointType.INCREMENTAL_CHECKPOINT) {
+ return snapshotIncrementally(checkpointId, timestamp,
streamFactory);
+ } else {
+ return snapshotFully(checkpointId, timestamp,
streamFactory);
+ }
+ }
+
+ private RunnableFuture<KeyedStateHandle> snapshotIncrementally(
+ final long checkpointId,
--- End diff --
This method has some amount of duplicated code with `snapshotFully`. My
suggestion would be target for some common abstract strategy common to all
checkpointing, e.g. `takeSnapshot`, `materialize`, and `releaseResources`. Then
we could have a factory that, depending on the checkpoint type, instantiates
the right implementation. This would also help with a second concern: I can see
that the code in this class has grown a lot over time, so this could be the
time to move some aspects like checkpointing strategies into separated classes
(they are already static inner classes anyways). By splitting a bit between
processing logic and snapshot/restore logic, we can keep things more modular
and separation of concerns.
We can also do this cleanup in a followup PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
