You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2025/02/19 12:13:44 UTC
(flink) branch release-2.0 updated: [FLINK-37335][state/forst] Handle the first full checkpoint for ForStIncrementalSnapshotStrategy (#26176)
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.0 by this push:
new b2a02f9b6fd [FLINK-37335][state/forst] Handle the first full checkpoint for ForStIncrementalSnapshotStrategy (#26176)
b2a02f9b6fd is described below
commit b2a02f9b6fd4afda0f6ccb0349b5249fad862b25
Author: AlexYinHan <yi...@alibaba-inc.com>
AuthorDate: Wed Feb 19 20:13:39 2025 +0800
[FLINK-37335][state/forst] Handle the first full checkpoint for ForStIncrementalSnapshotStrategy (#26176)
---
.../datatransfer/DataTransferStrategyBuilder.java | 12 ++++---
.../forst/datatransfer/ForStStateDataTransfer.java | 10 +++---
.../snapshot/ForStIncrementalSnapshotStrategy.java | 19 ++++++++---
.../snapshot/ForStNativeFullSnapshotStrategy.java | 6 ++--
.../forst/snapshot/ForStSnapshotStrategyBase.java | 4 +++
.../datatransfer/DataTransferStrategyTest.java | 6 ++--
.../datatransfer/ForStStateDataTransferTest.java | 18 +++++++----
.../ForStIncrementalSnapshotStrategyTest.java | 37 ++++++++++++++++++++--
8 files changed, 88 insertions(+), 24 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java
index fbbb68d113c..68d54b17799 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java
@@ -47,21 +47,25 @@ public class DataTransferStrategyBuilder {
public static DataTransferStrategy buildForSnapshot(
SnapshotType.SharingFilesStrategy sharingFilesStrategy,
@Nullable ForStFlinkFileSystem forStFlinkFileSystem,
- @Nullable CheckpointStreamFactory checkpointStreamFactory) {
+ @Nullable CheckpointStreamFactory checkpointStreamFactory,
+ boolean forceCopy) {
return buildForSnapshot(
sharingFilesStrategy,
forStFlinkFileSystem,
isDbPathUnderCheckpointPathForSnapshot(
- forStFlinkFileSystem, checkpointStreamFactory));
+ forStFlinkFileSystem, checkpointStreamFactory),
+ forceCopy);
}
@VisibleForTesting
static DataTransferStrategy buildForSnapshot(
SnapshotType.SharingFilesStrategy sharingFilesStrategy,
@Nullable ForStFlinkFileSystem forStFlinkFileSystem,
- boolean isDbPathUnderCheckpointPathForSnapshot) {
+ boolean isDbPathUnderCheckpointPathForSnapshot,
+ boolean forceCopy) {
DataTransferStrategy strategy;
- if (sharingFilesStrategy != SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD
+ if (forceCopy
+ || sharingFilesStrategy == SnapshotType.SharingFilesStrategy.NO_SHARING
|| forStFlinkFileSystem == null
|| !isDbPathUnderCheckpointPathForSnapshot) {
strategy =
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
index 61c4162f646..4dfc9c0a91f 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransfer.java
@@ -98,13 +98,14 @@ public class ForStStateDataTransfer implements Closeable {
CheckpointStreamFactory checkpointStreamFactory,
CheckpointedStateScope stateScope,
CloseableRegistry snapshotCloseableRegistry,
- CloseableRegistry tmpResourcesRegistry)
+ CloseableRegistry tmpResourcesRegistry,
+ boolean forceCopy)
throws Exception {
try {
DataTransferStrategy strategy =
DataTransferStrategyBuilder.buildForSnapshot(
- sharingFilesStrategy, forStFs, checkpointStreamFactory);
+ sharingFilesStrategy, forStFs, checkpointStreamFactory, forceCopy);
return createTransferFuture(
strategy,
file,
@@ -126,11 +127,12 @@ public class ForStStateDataTransfer implements Closeable {
CheckpointStreamFactory checkpointStreamFactory,
CheckpointedStateScope stateScope,
CloseableRegistry closeableRegistry,
- CloseableRegistry tmpResourcesRegistry)
+ CloseableRegistry tmpResourcesRegistry,
+ boolean forceCopy)
throws Exception {
DataTransferStrategy strategy =
DataTransferStrategyBuilder.buildForSnapshot(
- sharingFilesStrategy, forStFs, checkpointStreamFactory);
+ sharingFilesStrategy, forStFs, checkpointStreamFactory, forceCopy);
List<CompletableFuture<HandleAndLocalPath>> futures =
files.stream()
.map(
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategy.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategy.java
index 9d24710fe24..253381c801a 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategy.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategy.java
@@ -140,7 +140,15 @@ public class ForStIncrementalSnapshotStrategy<K> extends ForStNativeFullSnapshot
// savepoint, use empty PreviousSnapshot
snapshotResources.setPreviousSnapshot(EMPTY_PREVIOUS_SNAPSHOT);
break;
- case FORWARD: // full checkpoint for IncrementalSnapshotStrategy is not supported
+ case FORWARD:
+ // Full checkpoint for IncrementalSnapshotStrategy is not supported, except for the
+ // first one.
+ if (snapshotResources.previousSnapshot.isEmpty()) {
+ break;
+ } else {
+ throw new IllegalArgumentException(
+ "Triggering a full checkpoint for IncrementalSnapshotStrategy is not supported.");
+ }
default:
throw new IllegalArgumentException(
String.format(
@@ -338,7 +346,8 @@ public class ForStIncrementalSnapshotStrategy<K> extends ForStNativeFullSnapshot
checkpointStreamFactory,
stateScope,
snapshotCloseableRegistry,
- tmpResourcesRegistry);
+ tmpResourcesRegistry,
+ false);
sstHandles.addAll(sstFilesTransferResult);
transferBytes +=
@@ -353,7 +362,8 @@ public class ForStIncrementalSnapshotStrategy<K> extends ForStNativeFullSnapshot
checkpointStreamFactory,
stateScope,
snapshotCloseableRegistry,
- tmpResourcesRegistry);
+ tmpResourcesRegistry,
+ false);
metaHandles.addAll(miscFilesTransferResult);
transferBytes +=
miscFilesTransferResult.stream()
@@ -368,7 +378,8 @@ public class ForStIncrementalSnapshotStrategy<K> extends ForStNativeFullSnapshot
checkpointStreamFactory,
stateScope,
snapshotCloseableRegistry,
- tmpResourcesRegistry);
+ tmpResourcesRegistry,
+ false);
metaHandles.add(manifestFileTransferResult);
transferBytes += manifestFileTransferResult.getStateSize();
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStNativeFullSnapshotStrategy.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStNativeFullSnapshotStrategy.java
index 6baf92f0960..37674de9b86 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStNativeFullSnapshotStrategy.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStNativeFullSnapshotStrategy.java
@@ -321,7 +321,8 @@ public class ForStNativeFullSnapshotStrategy<K>
checkpointStreamFactory,
CheckpointedStateScope.EXCLUSIVE,
snapshotCloseableRegistry,
- tmpResourcesRegistry);
+ tmpResourcesRegistry,
+ true);
uploadedSize += uploadedFiles.stream().mapToLong(e -> e.getStateSize()).sum();
privateFiles.addAll(uploadedFiles);
@@ -333,7 +334,8 @@ public class ForStNativeFullSnapshotStrategy<K>
checkpointStreamFactory,
CheckpointedStateScope.EXCLUSIVE,
snapshotCloseableRegistry,
- tmpResourcesRegistry);
+ tmpResourcesRegistry,
+ true);
privateFiles.add(manifestFileTransferResult);
uploadedSize += manifestFileTransferResult.getStateSize();
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStSnapshotStrategyBase.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStSnapshotStrategyBase.java
index c72b4065ab4..4a9ddcdd49e 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStSnapshotStrategyBase.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStSnapshotStrategyBase.java
@@ -373,5 +373,9 @@ public abstract class ForStSnapshotStrategyBase<K, R extends SnapshotResources>
return Optional.empty();
}
}
+
+ protected boolean isEmpty() {
+ return confirmedSstFiles.isEmpty();
+ }
}
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java
index c91391d3913..459348e49b5 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java
@@ -381,7 +381,8 @@ public class DataTransferStrategyTest {
return DataTransferStrategyBuilder.buildForSnapshot(
SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
db.dbDelegateFileSystem,
- dbDirUnderCpDir);
+ dbDirUnderCpDir,
+ false);
}
private Tuple2<DBFilesContainer, DataTransferStrategy> createOrRestoreDb(
@@ -411,7 +412,8 @@ public class DataTransferStrategyTest {
DataTransferStrategyBuilder.buildForSnapshot(
sharingFilesStrategy,
forStFlinkFileSystem,
- isDbPathUnderCheckpointPathForSnapshot)
+ isDbPathUnderCheckpointPathForSnapshot,
+ false)
.getClass())
.isEqualTo(expected);
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransferTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransferTest.java
index a58f140d46c..bfa90404adf 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransferTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/datatransfer/ForStStateDataTransferTest.java
@@ -114,7 +114,8 @@ class ForStStateDataTransferTest extends TestLogger {
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
new CloseableRegistry(),
- new CloseableRegistry()))
+ new CloseableRegistry(),
+ false))
.isEqualTo(expectedException);
}
}
@@ -156,7 +157,8 @@ class ForStStateDataTransferTest extends TestLogger {
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
new CloseableRegistry(),
- tmpResourcesRegistry);
+ tmpResourcesRegistry,
+ false);
assertThatThrownBy(
() ->
@@ -169,7 +171,8 @@ class ForStStateDataTransferTest extends TestLogger {
expectedException),
CheckpointedStateScope.SHARED,
new CloseableRegistry(),
- tmpResourcesRegistry))
+ tmpResourcesRegistry,
+ false))
.isEqualTo(expectedException);
assertThat(checkpointPrivateFolder.list()).isEmpty();
assertThat(checkpointSharedFolder.list()).isNotEmpty();
@@ -188,7 +191,8 @@ class ForStStateDataTransferTest extends TestLogger {
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
new CloseableRegistry(),
- tmpResourcesRegistry))
+ tmpResourcesRegistry,
+ false))
.as("Cannot register Closeable, registry is already closed. Closing argument.")
.isInstanceOf(IOException.class);
@@ -233,7 +237,8 @@ class ForStStateDataTransferTest extends TestLogger {
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
new CloseableRegistry(),
- new CloseableRegistry());
+ new CloseableRegistry(),
+ false);
assertStateContentEqual(
sstFile, headBytes, handleAndLocalPath.getHandle().openInputStream());
@@ -275,7 +280,8 @@ class ForStStateDataTransferTest extends TestLogger {
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
new CloseableRegistry(),
- new CloseableRegistry());
+ new CloseableRegistry(),
+ false);
for (Path path : sstFilePaths) {
assertStateContentEqual(
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
index 6bb5a83e010..acd19b48b7e 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/snapshot/ForStIncrementalSnapshotStrategyTest.java
@@ -22,7 +22,9 @@ import org.apache.flink.api.common.state.v2.StateDescriptor;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.state.ArrayListSerializer;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.state.CompositeKeySerializationUtils;
import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
@@ -62,13 +64,27 @@ class ForStIncrementalSnapshotStrategyTest {
// Verify the next checkpoint is still incremental after a savepoint completed.
@Test
void testCheckpointIsIncremental() throws Exception {
+ testCheckpointIsIncremental(false);
+ }
+
+ @Test
+ void testIncrementalCheckpointWithFirstBeingFull() throws Exception {
+ testCheckpointIsIncremental(true);
+ }
+
+ private void testCheckpointIsIncremental(boolean forceFirstFull) throws Exception {
try (CloseableRegistry closeableRegistry = new CloseableRegistry();
ForStSnapshotStrategyBase checkpointSnapshotStrategy = createSnapshotStrategy()) {
FsCheckpointStreamFactory checkpointStreamFactory = createFsCheckpointStreamFactory();
// make and notify checkpoint with id 1
- snapshot(1L, checkpointSnapshotStrategy, checkpointStreamFactory, closeableRegistry);
+ snapshot(
+ 1L,
+ checkpointSnapshotStrategy,
+ checkpointStreamFactory,
+ closeableRegistry,
+ forceFirstFull);
checkpointSnapshotStrategy.notifyCheckpointComplete(1L);
// notify savepoint with id 2
@@ -201,7 +217,24 @@ class ForStIncrementalSnapshotStrategyTest {
FsCheckpointStreamFactory checkpointStreamFactory,
CloseableRegistry closeableRegistry)
throws Exception {
+ return snapshot(
+ checkpointId, snapshotStrategy, checkpointStreamFactory, closeableRegistry, false);
+ }
+
+ private <SR extends SnapshotResources> IncrementalRemoteKeyedStateHandle snapshot(
+ long checkpointId,
+ ForStSnapshotStrategyBase<?, SR> snapshotStrategy,
+ FsCheckpointStreamFactory checkpointStreamFactory,
+ CloseableRegistry closeableRegistry,
+ boolean forceFullCheckpoint)
+ throws Exception {
+ CheckpointOptions checkpointOptions =
+ new CheckpointOptions(
+ forceFullCheckpoint
+ ? CheckpointType.FULL_CHECKPOINT
+ : CheckpointType.CHECKPOINT,
+ CheckpointStorageLocationReference.getDefault());
return (IncrementalRemoteKeyedStateHandle)
snapshotStrategy
.asyncSnapshot(
@@ -209,7 +242,7 @@ class ForStIncrementalSnapshotStrategyTest {
checkpointId,
checkpointId,
checkpointStreamFactory,
- CheckpointOptions.forCheckpointWithDefaultLocation())
+ checkpointOptions)
.get(closeableRegistry)
.getJobManagerOwnedSnapshot();
}