This is an automated email from the ASF dual-hosted git repository.
zakelly 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 f975783ce9a [FLINK-37021][state/forst] Fix incorrect paths when
reusing and creating files. (#26040)
f975783ce9a is described below
commit f975783ce9a688a99e2f74fe6f2f5ad204589fdc
Author: AlexYinHan <[email protected]>
AuthorDate: Wed Feb 5 16:45:14 2025 +0800
[FLINK-37021][state/forst] Fix incorrect paths when reusing and creating
files. (#26040)
---
.../flink/state/forst/ForStResourceContainer.java | 15 ---
.../datatransfer/CopyDataTransferStrategy.java | 135 ++++++++++-----------
.../datatransfer/DataTransferStrategyBuilder.java | 41 ++++---
.../forst/datatransfer/ForStStateDataTransfer.java | 36 ++----
.../datatransfer/ReusableDataTransferStrategy.java | 43 ++-----
.../flink/state/forst/fs/ForStFlinkFileSystem.java | 29 +----
.../forst/fs/filemapping/FileMappingManager.java | 21 +---
.../forst/fs/filemapping/FileOwnershipDecider.java | 67 +---------
.../restore/ForStIncrementalRestoreOperation.java | 2 +-
.../snapshot/ForStIncrementalSnapshotStrategy.java | 11 +-
.../snapshot/ForStNativeFullSnapshotStrategy.java | 12 +-
.../state/forst/ForStResourceContainerTest.java | 2 +-
.../datatransfer/DataTransferStrategyTest.java | 119 +++++++++++++++---
.../datatransfer/ForStStateDataTransferTest.java | 7 ++
.../state/forst/fs/FileMappingManagerTest.java | 39 +-----
.../state/forst/fs/ForStFlinkFileSystemTest.java | 5 -
16 files changed, 252 insertions(+), 332 deletions(-)
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
index ba76b72371d..4f8a36dabb7 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
@@ -28,10 +28,8 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.memory.OpaqueMemoryResource;
import org.apache.flink.runtime.state.CheckpointStorageAccess;
-import org.apache.flink.state.forst.datatransfer.DataTransferStrategy;
import org.apache.flink.state.forst.fs.ForStFlinkFileSystem;
import org.apache.flink.state.forst.fs.StringifiedForStFileSystem;
-import org.apache.flink.state.forst.fs.filemapping.FileOwnershipDecider;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -106,12 +104,6 @@ public final class ForStResourceContainer implements
AutoCloseable {
/** The ForSt file system. Null when remote dir is not set. */
@Nullable private ForStFlinkFileSystem forStFileSystem;
- private final RecoveryClaimMode claimMode;
-
- private final FileOwnershipDecider fileOwnershipDecider;
-
- private DataTransferStrategy dataTransferStrategy;
-
/**
* The shared resource among ForSt instances. This resource is not part of
the 'handlesToClose',
* because the handles to close are closed quietly, whereas for this one,
we want exceptions to
@@ -193,8 +185,6 @@ public final class ForStResourceContainer implements
AutoCloseable {
this.remoteBasePath = remoteBasePath;
this.remoteForStPath =
remoteBasePath != null ? new Path(remoteBasePath,
DB_DIR_STRING) : null;
- this.claimMode = claimMode;
- this.fileOwnershipDecider = new FileOwnershipDecider(claimMode);
this.enableStatistics = enableStatistics;
this.handlesToClose = new ArrayList<>();
@@ -395,7 +385,6 @@ public final class ForStResourceContainer implements
AutoCloseable {
ForStFlinkFileSystem.get(
remoteForStPath.toUri(),
localForStPath,
- fileOwnershipDecider,
ForStFlinkFileSystem.getFileBasedCache(
cacheBasePath, cacheCapacity,
cacheReservedSize, metricGroup));
} else {
@@ -407,10 +396,6 @@ public final class ForStResourceContainer implements
AutoCloseable {
return forStFileSystem;
}
- public DataTransferStrategy getDataTransferStrategy() {
- return dataTransferStrategy;
- }
-
private static void prepareDirectories(Path basePath, Path dbPath) throws
IOException {
FileSystem fileSystem = basePath.getFileSystem();
if (fileSystem.exists(basePath)) {
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/CopyDataTransferStrategy.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/CopyDataTransferStrategy.java
index 3423318f8d8..860cbbc0234 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/CopyDataTransferStrategy.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/CopyDataTransferStrategy.java
@@ -31,10 +31,8 @@ import org.apache.flink.runtime.state.StateUtil;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.state.forst.fs.ForStFlinkFileSystem;
-import
org.apache.flink.state.forst.fs.filemapping.FileBackedMappingEntrySource;
-import
org.apache.flink.state.forst.fs.filemapping.HandleBackedMappingEntrySource;
+import org.apache.flink.state.forst.fs.filemapping.FileOwnership;
import org.apache.flink.state.forst.fs.filemapping.MappingEntry;
-import org.apache.flink.state.forst.fs.filemapping.MappingEntrySource;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -78,7 +76,6 @@ public class CopyDataTransferStrategy extends
DataTransferStrategy {
}
return copyFileToCheckpoint(
- dbFileSystem,
dbFilePath,
maxTransferBytes,
checkpointStreamFactory,
@@ -92,7 +89,7 @@ public class CopyDataTransferStrategy extends
DataTransferStrategy {
StreamStateHandle sourceHandle, Path targetPath, CloseableRegistry
closeableRegistry)
throws IOException {
LOG.trace("Copy file from checkpoint: {}, {}, {}", sourceHandle,
targetPath, dbFileSystem);
- copyFileFromCheckpoint(dbFileSystem, sourceHandle, targetPath,
closeableRegistry);
+ copyFileFromCheckpoint(sourceHandle, targetPath, closeableRegistry);
}
@Override
@@ -100,89 +97,86 @@ public class CopyDataTransferStrategy extends
DataTransferStrategy {
return "CopyDataTransferStrategy{" + ", dbFileSystem=" + dbFileSystem
+ '}';
}
- private static HandleAndLocalPath copyFileToCheckpoint(
- FileSystem dbFileSystem,
- Path filePath,
+ private HandleAndLocalPath copyFileToCheckpoint(
+ Path dbFilePath,
long maxTransferBytes,
CheckpointStreamFactory checkpointStreamFactory,
CheckpointedStateScope stateScope,
CloseableRegistry closeableRegistry,
CloseableRegistry tmpResourcesRegistry)
throws IOException {
- StreamStateHandle handleByDuplicating =
- duplicateFileToCheckpoint(
- dbFileSystem, filePath, checkpointStreamFactory,
stateScope);
- if (handleByDuplicating != null) {
- LOG.trace("Duplicate file to checkpoint: {} {}", filePath,
handleByDuplicating);
- return HandleAndLocalPath.of(handleByDuplicating,
filePath.getName());
+
+ // Get State handle for the DB file
+ StreamStateHandle sourceStateHandle;
+ if (dbFileSystem instanceof ForStFlinkFileSystem) {
+ // Obtain the state handle stored in MappingEntry
+ // or Construct a FileStateHandle base on the source file
+ MappingEntry mappingEntry =
+ ((ForStFlinkFileSystem)
dbFileSystem).getMappingEntry(dbFilePath);
+ Preconditions.checkNotNull(mappingEntry, "dbFile not found: " +
dbFilePath);
+ sourceStateHandle = mappingEntry.getSource().toStateHandle();
+ if (mappingEntry.getFileOwnership() == FileOwnership.NOT_OWNED) {
+ // The file is already owned by JM, simply return the state
handle
+ return HandleAndLocalPath.of(sourceStateHandle,
dbFilePath.getName());
+ }
+ } else {
+ // Construct a FileStateHandle base on the DB file
+ FileSystem sourceFileSystem = dbFilePath.getFileSystem();
+ long fileLength =
sourceFileSystem.getFileStatus(dbFilePath).getLen();
+ sourceStateHandle = new FileStateHandle(dbFilePath, fileLength);
}
- HandleAndLocalPath handleAndLocalPath =
- HandleAndLocalPath.of(
- writeFileToCheckpoint(
- dbFileSystem,
- filePath,
- maxTransferBytes,
- checkpointStreamFactory,
- stateScope,
- closeableRegistry,
- tmpResourcesRegistry),
- filePath.getName());
- LOG.trace("Write file to checkpoint: {}, {}", filePath,
handleAndLocalPath.getHandle());
- return handleAndLocalPath;
+ // Try path-copying first. If failed, fallback to bytes-copying
+ StreamStateHandle targetStateHandle =
+ tryPathCopyingToCheckpoint(sourceStateHandle,
checkpointStreamFactory, stateScope);
+ if (targetStateHandle != null) {
+ LOG.trace("Path-copy file to checkpoint: {} {}", dbFilePath,
targetStateHandle);
+ } else {
+ targetStateHandle =
+ bytesCopyingToCheckpoint(
+ dbFilePath,
+ maxTransferBytes,
+ checkpointStreamFactory,
+ stateScope,
+ closeableRegistry,
+ tmpResourcesRegistry);
+ LOG.trace("Bytes-copy file to checkpoint: {}, {}", dbFilePath,
targetStateHandle);
+ }
+
+ return HandleAndLocalPath.of(targetStateHandle, dbFilePath.getName());
}
/**
* Duplicate file to checkpoint storage by calling {@link
CheckpointStreamFactory#duplicate} if
* possible.
+ *
+ * @param sourceHandle The source state handle
+ * @param checkpointStreamFactory The checkpoint stream factory
+ * @param stateScope The state scope
+ * @return The target state handle if path-copying is successful,
otherwise null
*/
- private static @Nullable StreamStateHandle duplicateFileToCheckpoint(
- FileSystem dbFileSystem,
- Path filePath,
+ private @Nullable StreamStateHandle tryPathCopyingToCheckpoint(
+ @Nonnull StreamStateHandle sourceHandle,
CheckpointStreamFactory checkpointStreamFactory,
- CheckpointedStateScope stateScope)
- throws IOException {
-
- StreamStateHandle stateHandle = getStateHandle(dbFileSystem, filePath);
+ CheckpointedStateScope stateScope) {
- if (!checkpointStreamFactory.canFastDuplicate(stateHandle,
stateScope)) {
- return null;
- }
-
- List<StreamStateHandle> result =
- checkpointStreamFactory.duplicate(
- Collections.singletonList(stateHandle), stateScope);
- return result.get(0);
- }
-
- private static StreamStateHandle getStateHandle(FileSystem dbFileSystem,
Path filePath)
- throws IOException {
- Path sourceFilePath = filePath;
- if (dbFileSystem instanceof ForStFlinkFileSystem) {
- MappingEntry mappingEntry =
- ((ForStFlinkFileSystem)
dbFileSystem).getMappingEntry(filePath);
- Preconditions.checkNotNull(
- mappingEntry, "File mapping entry not found for %s",
filePath);
-
- MappingEntrySource source = mappingEntry.getSource();
- if (source instanceof HandleBackedMappingEntrySource) {
- // return the state handle stored in MappingEntry
- return ((HandleBackedMappingEntrySource)
source).getStateHandle();
- } else {
- // use file path stored in MappingEntry
- sourceFilePath = ((FileBackedMappingEntrySource)
source).getFilePath();
+ try {
+ if (!checkpointStreamFactory.canFastDuplicate(sourceHandle,
stateScope)) {
+ return null;
}
- }
- // construct a FileStateHandle base on source file
- FileSystem sourceFileSystem = sourceFilePath.getFileSystem();
- long fileLength =
sourceFileSystem.getFileStatus(sourceFilePath).getLen();
- return new FileStateHandle(sourceFilePath, fileLength);
+ List<StreamStateHandle> result =
+ checkpointStreamFactory.duplicate(
+ Collections.singletonList(sourceHandle),
stateScope);
+ return result.get(0);
+ } catch (Exception e) {
+ LOG.warn("Failed to duplicate file to checkpoint: {} {}",
sourceHandle, stateScope, e);
+ }
+ return null;
}
/** Write file to checkpoint storage through {@link
CheckpointStateOutputStream}. */
- private static @Nullable StreamStateHandle writeFileToCheckpoint(
- FileSystem dbFileSystem,
+ private @Nullable StreamStateHandle bytesCopyingToCheckpoint(
Path filePath,
long maxTransferBytes,
CheckpointStreamFactory checkpointStreamFactory,
@@ -236,11 +230,8 @@ public class CopyDataTransferStrategy extends
DataTransferStrategy {
}
}
- private static void copyFileFromCheckpoint(
- FileSystem dbFileSystem,
- StreamStateHandle sourceHandle,
- Path targetPath,
- CloseableRegistry closeableRegistry)
+ private void copyFileFromCheckpoint(
+ StreamStateHandle sourceHandle, Path targetPath, CloseableRegistry
closeableRegistry)
throws IOException {
if (closeableRegistry.isClosed()) {
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 d0cd3a4c527..7521f435615 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
@@ -21,16 +21,16 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.execution.RecoveryClaimMode;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.checkpoint.SnapshotType;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import org.apache.flink.runtime.state.IncrementalKeyedStateHandle;
import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
-import org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory;
+import org.apache.flink.runtime.state.filesystem.FsCheckpointStorageLocation;
import org.apache.flink.state.forst.StateHandleTransferSpec;
import org.apache.flink.state.forst.fs.ForStFlinkFileSystem;
-import org.apache.flink.state.forst.fs.filemapping.FileOwnershipDecider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,9 +45,11 @@ public class DataTransferStrategyBuilder {
private static final Logger LOG =
LoggerFactory.getLogger(DataTransferStrategyBuilder.class);
public static DataTransferStrategy buildForSnapshot(
+ SnapshotType.SharingFilesStrategy sharingFilesStrategy,
@Nullable ForStFlinkFileSystem forStFlinkFileSystem,
@Nullable CheckpointStreamFactory checkpointStreamFactory) {
return buildForSnapshot(
+ sharingFilesStrategy,
forStFlinkFileSystem,
isDbPathUnderCheckpointPathForSnapshot(
forStFlinkFileSystem, checkpointStreamFactory));
@@ -55,10 +57,13 @@ public class DataTransferStrategyBuilder {
@VisibleForTesting
static DataTransferStrategy buildForSnapshot(
+ SnapshotType.SharingFilesStrategy sharingFilesStrategy,
@Nullable ForStFlinkFileSystem forStFlinkFileSystem,
boolean isDbPathUnderCheckpointPathForSnapshot) {
DataTransferStrategy strategy;
- if (forStFlinkFileSystem == null ||
isDbPathUnderCheckpointPathForSnapshot) {
+ if (sharingFilesStrategy !=
SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD
+ || forStFlinkFileSystem == null
+ || !isDbPathUnderCheckpointPathForSnapshot) {
strategy =
forStFlinkFileSystem == null
? new CopyDataTransferStrategy()
@@ -67,9 +72,7 @@ public class DataTransferStrategyBuilder {
return strategy;
}
- strategy =
- new ReusableDataTransferStrategy(
- FileOwnershipDecider.getDefault(),
forStFlinkFileSystem);
+ strategy = new ReusableDataTransferStrategy(forStFlinkFileSystem);
LOG.info("Build DataTransferStrategy for Snapshot: {}", strategy);
return strategy;
}
@@ -80,17 +83,17 @@ public class DataTransferStrategyBuilder {
if (forStFlinkFileSystem == null) {
return false;
}
- // For checkpoint other that FsCheckpointStorageAccess, we treat it as
'DB path not under
+ // For checkpoint other than FsCheckpointStorageAccess, we treat it as
'DB path not under
// checkpoint path', since we cannot reuse checkpoint files in such
case.
// todo: Support enabling 'cp file reuse' with
FsMergingCheckpointStorageAccess
if (checkpointStreamFactory == null
- || checkpointStreamFactory.getClass() !=
FsCheckpointStreamFactory.class) {
+ || checkpointStreamFactory.getClass() !=
FsCheckpointStorageLocation.class) {
return false;
}
// get checkpoint shared path
- FsCheckpointStreamFactory fsCheckpointStreamFactory =
- (FsCheckpointStreamFactory) checkpointStreamFactory;
+ FsCheckpointStorageLocation fsCheckpointStreamFactory =
+ (FsCheckpointStorageLocation) checkpointStreamFactory;
Path cpSharedPath =
fsCheckpointStreamFactory.getTargetPath(CheckpointedStateScope.SHARED);
FileSystem cpSharedFs;
try {
@@ -101,13 +104,13 @@ public class DataTransferStrategyBuilder {
}
// check if dbRemotePath is under cpSharedPath
- if (!cpSharedFs.equals(forStFlinkFileSystem.getDelegateFS())) {
+ if
(!cpSharedFs.getUri().equals(forStFlinkFileSystem.getDelegateFS().getUri())) {
// different file system
return false;
} else {
// same file system
String dbRemotePathStr = forStFlinkFileSystem.getRemoteBase();
- String cpSharedPathStr = cpSharedPath.getPath();
+ String cpSharedPathStr = cpSharedPath.toString();
return cpSharedPathStr.equals(dbRemotePathStr)
|| dbRemotePathStr.startsWith(cpSharedPathStr);
}
@@ -121,18 +124,22 @@ public class DataTransferStrategyBuilder {
FileSystem cpSharedFs = getSharedStateFileSystem(specs);
if (forStFlinkFileSystem == null
|| cpSharedFs == null
- || forStFlinkFileSystem.equals(cpSharedFs)) {
+ || !forStFlinkFileSystem.getUri().equals(cpSharedFs.getUri())
+ || recoveryClaimMode != RecoveryClaimMode.CLAIM) {
strategy =
forStFlinkFileSystem == null
? new CopyDataTransferStrategy()
: new
CopyDataTransferStrategy(forStFlinkFileSystem);
- LOG.info("Build DataTransferStrategy for Restore: {}", strategy);
+ LOG.info(
+ "Build DataTransferStrategy for Restore: {},
forStFlinkFileSystem: {}, cpSharedFs:{}, recoveryClaimMode:{}",
+ strategy,
+ forStFlinkFileSystem,
+ cpSharedFs,
+ recoveryClaimMode);
return strategy;
}
- strategy =
- new ReusableDataTransferStrategy(
- new FileOwnershipDecider(recoveryClaimMode),
forStFlinkFileSystem);
+ strategy = new ReusableDataTransferStrategy(forStFlinkFileSystem);
LOG.info("Build DataTransferStrategy for Restore: {}", strategy);
return 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 61a2a121525..61c4162f646 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
@@ -21,6 +21,7 @@ import org.apache.flink.core.execution.RecoveryClaimMode;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.checkpoint.SnapshotType;
import org.apache.flink.runtime.state.CheckpointStateOutputStream;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointedStateScope;
@@ -91,6 +92,7 @@ public class ForStStateDataTransfer implements Closeable {
* will be transferred.
*/
public HandleAndLocalPath transferFileToCheckpointFs(
+ SnapshotType.SharingFilesStrategy sharingFilesStrategy,
Path file,
long transferBytes,
CheckpointStreamFactory checkpointStreamFactory,
@@ -100,7 +102,11 @@ public class ForStStateDataTransfer implements Closeable {
throws Exception {
try {
+ DataTransferStrategy strategy =
+ DataTransferStrategyBuilder.buildForSnapshot(
+ sharingFilesStrategy, forStFs,
checkpointStreamFactory);
return createTransferFuture(
+ strategy,
file,
transferBytes,
checkpointStreamFactory,
@@ -115,18 +121,22 @@ public class ForStStateDataTransfer implements Closeable {
/** Transfer a batch of files to checkpoint filesystem. */
public List<HandleAndLocalPath> transferFilesToCheckpointFs(
+ SnapshotType.SharingFilesStrategy sharingFilesStrategy,
List<Path> files,
CheckpointStreamFactory checkpointStreamFactory,
CheckpointedStateScope stateScope,
CloseableRegistry closeableRegistry,
CloseableRegistry tmpResourcesRegistry)
throws Exception {
-
+ DataTransferStrategy strategy =
+ DataTransferStrategyBuilder.buildForSnapshot(
+ sharingFilesStrategy, forStFs,
checkpointStreamFactory);
List<CompletableFuture<HandleAndLocalPath>> futures =
files.stream()
.map(
file ->
createTransferFuture(
+ strategy,
file,
-1,
checkpointStreamFactory,
@@ -188,6 +198,7 @@ public class ForStStateDataTransfer implements Closeable {
}
private CompletableFuture<HandleAndLocalPath> createTransferFuture(
+ DataTransferStrategy strategy,
Path file,
long transferBytes,
CheckpointStreamFactory checkpointStreamFactory,
@@ -197,7 +208,7 @@ public class ForStStateDataTransfer implements Closeable {
return CompletableFuture.supplyAsync(
CheckedSupplier.unchecked(
() ->
- transferFile(
+ strategy.transferToCheckpoint(
file,
transferBytes,
checkpointStreamFactory,
@@ -211,27 +222,6 @@ public class ForStStateDataTransfer implements Closeable {
return forStFs != null ? forStFs : FileSystem.getLocalFileSystem();
}
- private HandleAndLocalPath transferFile(
- Path filePath,
- long maxTransferBytes,
- CheckpointStreamFactory checkpointStreamFactory,
- CheckpointedStateScope stateScope,
- CloseableRegistry closeableRegistry,
- CloseableRegistry tmpResourcesRegistry)
- throws IOException {
-
- DataTransferStrategy strategy =
- DataTransferStrategyBuilder.buildForSnapshot(forStFs,
checkpointStreamFactory);
-
- return strategy.transferToCheckpoint(
- filePath,
- maxTransferBytes,
- checkpointStreamFactory,
- stateScope,
- closeableRegistry,
- tmpResourcesRegistry);
- }
-
/**
* Transfer all state data to the target directory, as specified in the
transfer requests.
*
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ReusableDataTransferStrategy.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ReusableDataTransferStrategy.java
index beca54ba434..dff3abe36c8 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ReusableDataTransferStrategy.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/datatransfer/ReusableDataTransferStrategy.java
@@ -24,8 +24,6 @@ import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import
org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath;
import org.apache.flink.runtime.state.StreamStateHandle;
-import org.apache.flink.runtime.state.filemerging.SegmentFileStateHandle;
-import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
import org.apache.flink.state.forst.fs.ForStFlinkFileSystem;
import org.apache.flink.state.forst.fs.filemapping.FileOwnership;
@@ -34,9 +32,6 @@ import
org.apache.flink.state.forst.fs.filemapping.MappingEntry;
import org.apache.flink.state.forst.fs.filemapping.MappingEntrySource;
import org.apache.flink.util.Preconditions;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.Objects;
@@ -46,10 +41,7 @@ import java.util.Objects;
*/
public class ReusableDataTransferStrategy extends CopyDataTransferStrategy {
- private final FileOwnershipDecider fileOwnershipDecider;
-
- ReusableDataTransferStrategy(
- @Nonnull FileOwnershipDecider fileOwnershipDecider, FileSystem
dbFileSystem) {
+ ReusableDataTransferStrategy(FileSystem dbFileSystem) {
super(dbFileSystem);
Preconditions.checkArgument(
@@ -58,8 +50,6 @@ public class ReusableDataTransferStrategy extends
CopyDataTransferStrategy {
+ dbFileSystem.getClass()
+ ", expected: "
+ ForStFlinkFileSystem.class);
-
- this.fileOwnershipDecider = fileOwnershipDecider;
}
private ForStFlinkFileSystem getForStFlinkFileSystem() {
@@ -67,19 +57,15 @@ public class ReusableDataTransferStrategy extends
CopyDataTransferStrategy {
return (ForStFlinkFileSystem) dbFileSystem;
}
- private @Nullable HandleAndLocalPath reuseFileToCheckpoint(
- Path dbFilePath, ForStFlinkFileSystem forStFs) throws IOException {
+ private HandleAndLocalPath reuseFileToCheckpoint(Path dbFilePath,
ForStFlinkFileSystem forStFs)
+ throws IOException {
LOG.trace("Reuse file to checkpoint: {}", dbFilePath);
// Find the real path of the file
- Path sourcePath = forStFs.srcPath(dbFilePath);
- if (sourcePath != null) {
- dbFilePath = sourcePath;
- }
+ MappingEntry mappingEntry =
Objects.requireNonNull(forStFs.getMappingEntry(dbFilePath));
+ MappingEntrySource source = mappingEntry.getSource();
// Create a StateHandle with the real path or the restored handle
- MappingEntrySource source =
-
Objects.requireNonNull(forStFs.getMappingEntry(dbFilePath)).getSource();
StreamStateHandle stateHandle = source.toStateHandle();
// Give file ownership to JM, i.e. DB will not delete it from now on
@@ -134,7 +120,7 @@ public class ReusableDataTransferStrategy extends
CopyDataTransferStrategy {
return;
}
- FileOwnership fileOwnership =
fileOwnershipDecider.decideForRestoredFile(targetPath);
+ FileOwnership fileOwnership =
FileOwnershipDecider.decideForRestoredFile(targetPath);
if (fileOwnership == FileOwnership.PRIVATE_OWNED_BY_DB) {
LOG.trace(
"Not reusing file from checkpoint because the file is
privately owned by DB: {}",
@@ -145,16 +131,6 @@ public class ReusableDataTransferStrategy extends
CopyDataTransferStrategy {
reuseFileFromCheckpoint(sourceHandle, targetPath);
}
- private @Nullable Path prepareSourcePathFromStateHandle(StreamStateHandle
sourceHandle) {
- Path sourcePath = null;
- if (sourceHandle instanceof FileStateHandle) {
- sourcePath = ((FileStateHandle) sourceHandle).getFilePath();
- } else if (sourceHandle instanceof SegmentFileStateHandle) {
- sourcePath = ((SegmentFileStateHandle) sourceHandle).getFilePath();
- }
- return sourcePath;
- }
-
private void reuseFileFromCheckpoint(StreamStateHandle sourceHandle, Path
targetPath)
throws IOException {
LOG.trace("Reuse file from checkpoint: {}, {}", sourceHandle,
targetPath);
@@ -168,11 +144,6 @@ public class ReusableDataTransferStrategy extends
CopyDataTransferStrategy {
@Override
public String toString() {
- return "ReusableDataTransferStrategy{"
- + "dbFileSystem="
- + dbFileSystem
- + ", fileOwnershipDecider="
- + fileOwnershipDecider
- + '}';
+ return "ReusableDataTransferStrategy{" + "dbFileSystem=" +
dbFileSystem + '}';
}
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
index 6efda380aa9..045c3a81d4c 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
@@ -80,14 +80,12 @@ public class ForStFlinkFileSystem extends FileSystem {
FileSystem delegateFS,
String remoteBase,
String localBase,
- @Nullable FileOwnershipDecider fileOwnershipDecider,
@Nullable FileBasedCache fileBasedCache) {
this.localFS = FileSystem.getLocalFileSystem();
this.delegateFS = delegateFS;
this.remoteBase = remoteBase;
this.fileBasedCache = fileBasedCache;
- this.fileMappingManager =
- new FileMappingManager(delegateFS, fileOwnershipDecider,
remoteBase, localBase);
+ this.fileMappingManager = new FileMappingManager(delegateFS,
remoteBase, localBase);
}
/**
@@ -101,26 +99,14 @@ public class ForStFlinkFileSystem extends FileSystem {
*/
public static ForStFlinkFileSystem get(URI uri) throws IOException {
return new ForStFlinkFileSystem(
- FileSystem.get(uri),
- uri.toString(),
- System.getProperty("java.io.tmpdir"),
- null,
- null);
+ FileSystem.get(uri), uri.toString(),
System.getProperty("java.io.tmpdir"), null);
}
- public static ForStFlinkFileSystem get(
- URI uri,
- Path localBase,
- FileOwnershipDecider fileOwnershipDecider,
- FileBasedCache fileBasedCache)
+ public static ForStFlinkFileSystem get(URI uri, Path localBase,
FileBasedCache fileBasedCache)
throws IOException {
Preconditions.checkNotNull(localBase, "localBase is null, remote uri:
%s.", uri);
return new ForStFlinkFileSystem(
- FileSystem.get(uri),
- uri.toString(),
- localBase.toString(),
- fileOwnershipDecider,
- fileBasedCache);
+ FileSystem.get(uri), uri.toString(), localBase.toString(),
fileBasedCache);
}
public static FileBasedCache getFileBasedCache(
@@ -193,7 +179,7 @@ public class ForStFlinkFileSystem extends FileSystem {
CachedDataOutputStream cachedDataOutputStream =
createCachedDataOutputStream(dbFilePath, sourceRealPath,
outputStream);
- LOG.info(
+ LOG.trace(
"Create file: dbFilePath: {}, sourceRealPath: {},
cachedDataOutputStream: {}",
dbFilePath,
sourceRealPath,
@@ -354,11 +340,6 @@ public class ForStFlinkFileSystem extends FileSystem {
fileMappingManager.registerReusedRestoredFile(key, stateHandle,
dbFilePath);
}
- public synchronized @Nullable Path srcPath(Path path) {
- MappingEntry mappingEntry =
fileMappingManager.mappingEntry(path.toString());
- return mappingEntry == null ? null : mappingEntry.getSourcePath();
- }
-
public synchronized @Nullable MappingEntry getMappingEntry(Path path) {
return fileMappingManager.mappingEntry(path.toString());
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java
index 155c745714c..7d5544e8f8d 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileMappingManager.java
@@ -50,21 +50,11 @@ public class FileMappingManager {
private final HashMap<String, MappingEntry> mappingTable;
- private final FileOwnershipDecider fileOwnershipDecider;
-
private final String remoteBase;
private final String localBase;
- public FileMappingManager(
- FileSystem fileSystem,
- @Nullable FileOwnershipDecider fileOwnershipDecider,
- String remoteBase,
- String localBase) {
+ public FileMappingManager(FileSystem fileSystem, String remoteBase, String
localBase) {
this.fileSystem = fileSystem;
- this.fileOwnershipDecider =
- fileOwnershipDecider == null
- ? FileOwnershipDecider.getDefault()
- : fileOwnershipDecider;
this.mappingTable = new HashMap<>();
this.remoteBase = remoteBase;
this.localBase = localBase;
@@ -77,11 +67,8 @@ public class FileMappingManager {
filePath = forceLocalPath(filePath);
}
- filePath = toUUIDPath(filePath);
-
- LOG.trace("decide file ownership for new file: {} {}", filePath,
fileOwnershipDecider);
return addFileToMappingTable(
- key, filePath,
fileOwnershipDecider.decideForNewFile(filePath));
+ key, toUUIDPath(filePath),
FileOwnershipDecider.decideForNewFile(filePath));
}
/** Register a file restored from checkpoints to the mapping table. */
@@ -92,7 +79,7 @@ public class FileMappingManager {
// - Add to mapping table based on cpFilePath, so we can access the
real file.
LOG.trace("decide restored file ownership based on dbFilePath: {}",
dbFilePath);
return addHandleBackedFileToMappingTable(
- key, stateHandle,
fileOwnershipDecider.decideForRestoredFile(dbFilePath));
+ key, stateHandle,
FileOwnershipDecider.decideForRestoredFile(dbFilePath));
}
private MappingEntry addHandleBackedFileToMappingTable(
@@ -109,7 +96,7 @@ public class FileMappingManager {
MappingEntrySource source = new FileBackedMappingEntrySource(filePath);
MappingEntry existingEntry = getExistingMappingEntry(key, source,
fileOwnership);
return existingEntry == null
- ? addMappingEntry(key, new MappingEntry(1, filePath,
fileOwnership, false))
+ ? addMappingEntry(key, new MappingEntry(1, source,
fileOwnership, false))
: existingEntry;
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileOwnershipDecider.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileOwnershipDecider.java
index 75763f54910..909b4476270 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileOwnershipDecider.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/filemapping/FileOwnershipDecider.java
@@ -17,78 +17,22 @@
package org.apache.flink.state.forst.fs.filemapping;
-import org.apache.flink.core.execution.RecoveryClaimMode;
-import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.state.CheckpointStorageAccess;
-import org.apache.flink.runtime.state.filesystem.FsCheckpointStorageAccess;
-import
org.apache.flink.runtime.state.filesystem.FsMergingCheckpointStorageAccess;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
public class FileOwnershipDecider {
public static final String SST_SUFFIX = ".sst";
- private final RecoveryClaimMode recoveryClaimMode;
-
- public static FileOwnershipDecider getDefault() {
- return new FileOwnershipDecider(RecoveryClaimMode.DEFAULT);
- }
-
- private static boolean isDbPathUnderCheckpointPath(
- @Nullable CheckpointStorageAccess checkpointStorageAccess,
- @Nullable Path dbRemotePath) {
- if (dbRemotePath == null) {
- return false;
- }
-
- // For checkpoint other that FsCheckpointStorageAccess, we treat it as
'DB path not under
- // checkpoint path', since we cannot reuse checkpoint files in such
case.
- // todo: Support enabling 'cp file reuse' with
FsMergingCheckpointStorageAccess
- if (!(checkpointStorageAccess instanceof FsCheckpointStorageAccess)
- || checkpointStorageAccess instanceof
FsMergingCheckpointStorageAccess) {
- return false;
- }
-
- FsCheckpointStorageAccess fsCheckpointStorageAccess =
- (FsCheckpointStorageAccess) checkpointStorageAccess;
- FileSystem checkpointFS = fsCheckpointStorageAccess.getFileSystem();
- FileSystem dbRemoteFS;
- try {
- dbRemoteFS = dbRemotePath.getFileSystem();
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to get FileSystem from dbRemotePath: " +
dbRemotePath, e);
- }
+ private FileOwnershipDecider() {}
- if (!checkpointFS.equals(dbRemoteFS)) {
- // different file system
- return false;
- } else {
- // same file system
- String checkpointPathStr =
-
fsCheckpointStorageAccess.getCheckpointsDirectory().getPath();
- String dbRemotePathStr = dbRemotePath.getPath();
- return checkpointPathStr.equals(dbRemotePathStr)
- || dbRemotePathStr.startsWith(checkpointPathStr);
- }
- }
-
- public FileOwnershipDecider(RecoveryClaimMode recoveryClaimMode) {
- this.recoveryClaimMode = recoveryClaimMode;
- }
-
- public FileOwnership decideForNewFile(Path filePath) {
+ public static FileOwnership decideForNewFile(Path filePath) {
// local files are always privately owned by DB
return shouldAlwaysBeLocal(filePath)
? FileOwnership.PRIVATE_OWNED_BY_DB
: FileOwnership.SHAREABLE_OWNED_BY_DB;
}
- public FileOwnership decideForRestoredFile(Path filePath) {
+ public static FileOwnership decideForRestoredFile(Path filePath) {
// local files are always privately owned by DB
return shouldAlwaysBeLocal(filePath)
? FileOwnership.PRIVATE_OWNED_BY_DB
@@ -102,9 +46,4 @@ public class FileOwnershipDecider {
public static boolean shouldAlwaysBeLocal(Path filePath) {
return !isSstFile(filePath);
}
-
- @Override
- public String toString() {
- return "FileOwnershipDecider{" + "recoveryClaimMode=" +
recoveryClaimMode + '}';
- }
}
diff --git
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
index b69f64c029d..d3933e7ec82 100644
---
a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
+++
b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/restore/ForStIncrementalRestoreOperation.java
@@ -263,7 +263,7 @@ public class ForStIncrementalRestoreOperation<K> implements
ForStRestoreOperatio
ForStStateDataTransfer.DEFAULT_THREAD_NUM,
optionsContainer.getFileSystem())) {
transfer.transferAllStateDataToDirectory(
- specs, cancelStreamRegistry, RecoveryClaimMode.DEFAULT);
+ specs, cancelStreamRegistry, recoveryClaimMode);
}
}
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 13d3e8bbba2..9d24710fe24 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
@@ -136,14 +136,16 @@ public class ForStIncrementalSnapshotStrategy<K> extends
ForStNativeFullSnapshot
case FORWARD_BACKWARD:
// incremental checkpoint, use origin PreviousSnapshot
break;
- case FORWARD:
case NO_SHARING:
- // full checkpoint, use empty PreviousSnapshot
+ // savepoint, use empty PreviousSnapshot
snapshotResources.setPreviousSnapshot(EMPTY_PREVIOUS_SNAPSHOT);
break;
+ case FORWARD: // full checkpoint for IncrementalSnapshotStrategy
is not supported
default:
throw new IllegalArgumentException(
- "Unsupported sharing files strategy: " +
sharingFilesStrategy);
+ String.format(
+ "Unsupported sharing files strategy for %s :
%s",
+ this.getClass().getName(),
sharingFilesStrategy));
}
return new ForStIncrementalSnapshotOperation(
@@ -331,6 +333,7 @@ public class ForStIncrementalSnapshotStrategy<K> extends
ForStNativeFullSnapshot
List<HandleAndLocalPath> sstFilesTransferResult =
stateTransfer.transferFilesToCheckpointFs(
+ sharingFilesStrategy,
classifiedFiles.f1,
checkpointStreamFactory,
stateScope,
@@ -345,6 +348,7 @@ public class ForStIncrementalSnapshotStrategy<K> extends
ForStNativeFullSnapshot
List<HandleAndLocalPath> miscFilesTransferResult =
stateTransfer.transferFilesToCheckpointFs(
+ sharingFilesStrategy,
classifiedFiles.f2,
checkpointStreamFactory,
stateScope,
@@ -358,6 +362,7 @@ public class ForStIncrementalSnapshotStrategy<K> extends
ForStNativeFullSnapshot
HandleAndLocalPath manifestFileTransferResult =
stateTransfer.transferFileToCheckpointFs(
+ sharingFilesStrategy,
classifiedFiles.f3,
snapshotResources.manifestFileSize,
checkpointStreamFactory,
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 a47bbb568ee..6baf92f0960 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
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.SnapshotType;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import org.apache.flink.runtime.state.IncrementalKeyedStateHandle;
@@ -228,20 +229,27 @@ public class ForStNativeFullSnapshotStrategy<K>
if (syncPartResource.stateMetaInfoSnapshots.isEmpty()) {
return registry -> SnapshotResult.empty();
}
- return new ForStNativeFullSnapshotOperation(checkpointId,
streamFactory, syncPartResource);
+ return new ForStNativeFullSnapshotOperation(
+
checkpointOptions.getCheckpointType().getSharingFilesStrategy(),
+ checkpointId,
+ streamFactory,
+ syncPartResource);
}
/** Encapsulates the process to perform a full snapshot of a
ForStKeyedStateBackend. */
private final class ForStNativeFullSnapshotOperation extends
ForStSnapshotOperation {
@Nonnull private final ForStNativeSnapshotResources snapshotResources;
+ @Nonnull private final SnapshotType.SharingFilesStrategy
sharingFilesStrategy;
private ForStNativeFullSnapshotOperation(
+ SnapshotType.SharingFilesStrategy sharingFilesStrategy,
long checkpointId,
@Nonnull CheckpointStreamFactory checkpointStreamFactory,
@Nonnull ForStNativeSnapshotResources snapshotResources) {
super(checkpointId, checkpointStreamFactory);
this.snapshotResources = snapshotResources;
+ this.sharingFilesStrategy = sharingFilesStrategy;
}
@Override
@@ -301,6 +309,7 @@ public class ForStNativeFullSnapshotStrategy<K>
if (snapshotResources.liveFiles.size() > 0) {
List<IncrementalKeyedStateHandle.HandleAndLocalPath>
uploadedFiles =
stateTransfer.transferFilesToCheckpointFs(
+ sharingFilesStrategy,
snapshotResources.liveFiles.stream()
.filter(
file ->
@@ -318,6 +327,7 @@ public class ForStNativeFullSnapshotStrategy<K>
IncrementalKeyedStateHandle.HandleAndLocalPath
manifestFileTransferResult =
stateTransfer.transferFileToCheckpointFs(
+ SnapshotType.SharingFilesStrategy.NO_SHARING,
snapshotResources.manifestFilePath,
snapshotResources.manifestFileSize,
checkpointStreamFactory,
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
index 5d4ccde4736..3a33a645a97 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStResourceContainerTest.java
@@ -347,7 +347,7 @@ public class ForStResourceContainerTest {
DBOptions dbOptions2 =
new
DBOptions().setCreateIfMissing(true).setAvoidFlushDuringShutdown(true);
ForStFlinkFileSystem fileSystem =
- ForStFlinkFileSystem.get(remoteBasePath.toUri(),
localBasePath, null, null);
+ ForStFlinkFileSystem.get(remoteBasePath.toUri(),
localBasePath, null);
dbOptions2.setEnv(
new FlinkEnv(
remoteBasePath.toString(), new
StringifiedForStFileSystem(fileSystem)));
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 75748c1ecbe..7b06d74d94b 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
@@ -26,12 +26,16 @@ import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.local.LocalFileSystem;
+import org.apache.flink.runtime.checkpoint.SnapshotType;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import
org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath;
+import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle;
+import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory;
+import org.apache.flink.state.forst.StateHandleTransferSpec;
import org.apache.flink.state.forst.fs.ForStFlinkFileSystem;
import org.apache.flink.state.forst.fs.filemapping.FileOwnershipDecider;
import
org.apache.flink.state.forst.fs.filemapping.HandleBackedMappingEntrySource;
@@ -44,18 +48,23 @@ import org.apache.flink.util.FileUtils;
import org.apache.flink.util.Preconditions;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.io.TempDir;
+import javax.annotation.Nullable;
+
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -102,9 +111,7 @@ public class DataTransferStrategyTest {
FileOwnershipDecider fileOwnershipDecider;
- DBFilesContainer(
- Path dbLocalBase, Path dbRemoteBase, FileOwnershipDecider
fileOwnershipDecider)
- throws IOException {
+ DBFilesContainer(Path dbLocalBase, Path dbRemoteBase) throws
IOException {
realFileSystem = LocalFileSystem.getLocalFileSystem();
// prepare db paths
@@ -358,11 +365,7 @@ public class DataTransferStrategyTest {
private static final long MAX_TRANSFER_BYTES = Long.MAX_VALUE;
private DBFilesContainer createDb(
- JobID jobID,
- int subtaskIndex,
- int subtaskParallelism,
- boolean dbDirUnderCpDir,
- FileOwnershipDecider fileOwnershipDecider)
+ JobID jobID, int subtaskIndex, int subtaskParallelism, boolean
dbDirUnderCpDir)
throws IOException {
String dbIdentifier = String.format("%s-db-%d-%d", jobID,
subtaskIndex, subtaskParallelism);
Path dbLocalBase = new Path(tempDir.toString(),
String.format("local/%s", dbIdentifier));
@@ -372,16 +375,17 @@ public class DataTransferStrategyTest {
dbDirUnderCpDir
? String.format("checkpoint/%s", dbIdentifier)
: String.format("remote/%s", dbIdentifier));
- DBFilesContainer db = new DBFilesContainer(dbLocalBase, dbRemoteBase,
fileOwnershipDecider);
+ DBFilesContainer db = new DBFilesContainer(dbLocalBase, dbRemoteBase);
db.clear();
return db;
}
- private DataTransferStrategy createDataTransferStrategy(
- DBFilesContainer db, RecoveryClaimMode claimMode) {
+ private DataTransferStrategy createDataTransferStrategy(DBFilesContainer
db) {
return DataTransferStrategyBuilder.buildForSnapshot(
- db.dbDelegateFileSystem, dbDirUnderCpDir);
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
+ db.dbDelegateFileSystem,
+ dbDirUnderCpDir);
}
private Tuple2<DBFilesContainer, DataTransferStrategy> createOrRestoreDb(
@@ -391,14 +395,8 @@ public class DataTransferStrategyTest {
boolean dbDirUnderCpDir,
RecoveryClaimMode claimMode)
throws IOException {
- DBFilesContainer db =
- createDb(
- jobID,
- subtaskIndex,
- subtaskParallelism,
- dbDirUnderCpDir,
- new FileOwnershipDecider(claimMode));
- DataTransferStrategy strategy = createDataTransferStrategy(db,
claimMode);
+ DBFilesContainer db = createDb(jobID, subtaskIndex,
subtaskParallelism, dbDirUnderCpDir);
+ DataTransferStrategy strategy = createDataTransferStrategy(db);
return new Tuple2<>(db, strategy);
}
@@ -408,6 +406,87 @@ public class DataTransferStrategyTest {
FileUtils.cleanDirectory(tempDir.toFile());
}
+ void testSnapshotStrategyAsExpected(
+ SnapshotType.SharingFilesStrategy sharingFilesStrategy,
+ @Nullable ForStFlinkFileSystem forStFlinkFileSystem,
+ boolean isDbPathUnderCheckpointPathForSnapshot,
+ Class<?> expected) {
+ assertThat(
+ DataTransferStrategyBuilder.buildForSnapshot(
+ sharingFilesStrategy,
+ forStFlinkFileSystem,
+ isDbPathUnderCheckpointPathForSnapshot)
+ .getClass())
+ .isEqualTo(expected);
+ }
+
+ void testRestoreStrategyAsExpected(
+ @Nullable ForStFlinkFileSystem forStFlinkFileSystem,
+ RecoveryClaimMode recoveryClaimMode,
+ Class<?> expected) {
+ List<HandleAndLocalPath> sharedStateHandleList = new ArrayList<>();
+ sharedStateHandleList.add(
+ HandleAndLocalPath.of(new FileStateHandle(new Path("1.sst"),
0), "1.sst"));
+ IncrementalRemoteKeyedStateHandle stateHandle =
+ new IncrementalRemoteKeyedStateHandle(
+ UUID.randomUUID(),
+ KeyGroupRange.of(0, 0),
+ 0L,
+ sharedStateHandleList,
+ Collections.emptyList(),
+ new FileStateHandle(new Path("meta"), 0));
+ assertThat(
+ DataTransferStrategyBuilder.buildForRestore(
+ forStFlinkFileSystem,
+ Collections.singletonList(
+ new StateHandleTransferSpec(
+ stateHandle, new
Path("dst"))),
+ recoveryClaimMode)
+ .getClass())
+ .isEqualTo(expected);
+ }
+
+ @Test
+ void testBuildingStrategyAsExpected() throws IOException {
+ ForStFlinkFileSystem forStFlinkFileSystem =
ForStFlinkFileSystem.get(tempDir.toUri());
+
+ testSnapshotStrategyAsExpected(
+ SnapshotType.SharingFilesStrategy.NO_SHARING,
+ forStFlinkFileSystem,
+ true,
+ CopyDataTransferStrategy.class);
+
+ testSnapshotStrategyAsExpected(
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
+ forStFlinkFileSystem,
+ true,
+ ReusableDataTransferStrategy.class);
+
+ testSnapshotStrategyAsExpected(
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
+ forStFlinkFileSystem,
+ false,
+ CopyDataTransferStrategy.class);
+
+ testSnapshotStrategyAsExpected(
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
+ null,
+ false,
+ CopyDataTransferStrategy.class);
+
+ testRestoreStrategyAsExpected(
+ forStFlinkFileSystem, RecoveryClaimMode.CLAIM,
ReusableDataTransferStrategy.class);
+
+ testRestoreStrategyAsExpected(
+ forStFlinkFileSystem, RecoveryClaimMode.NO_CLAIM,
CopyDataTransferStrategy.class);
+
+ testRestoreStrategyAsExpected(
+ null, RecoveryClaimMode.CLAIM, CopyDataTransferStrategy.class);
+
+ testRestoreStrategyAsExpected(
+ null, RecoveryClaimMode.NO_CLAIM,
CopyDataTransferStrategy.class);
+ }
+
@TestTemplate
void simpleCaseTestRestore() throws IOException {
JobID jobID = new JobID();
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 c226b63a6bd..a58f140d46c 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
@@ -24,6 +24,7 @@ import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.checkpoint.SnapshotType;
import org.apache.flink.runtime.state.CheckpointStateOutputStream;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointedStateScope;
@@ -108,6 +109,7 @@ class ForStStateDataTransferTest extends TestLogger {
assertThatThrownBy(
() ->
stateTransfer.transferFilesToCheckpointFs(
+
SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
filePaths,
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
@@ -149,6 +151,7 @@ class ForStStateDataTransferTest extends TestLogger {
CloseableRegistry tmpResourcesRegistry = new CloseableRegistry();
try (ForStStateDataTransfer stateTransfer = new
ForStStateDataTransfer(1)) {
stateTransfer.transferFilesToCheckpointFs(
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
filePaths,
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
@@ -158,6 +161,7 @@ class ForStStateDataTransferTest extends TestLogger {
assertThatThrownBy(
() ->
stateTransfer.transferFilesToCheckpointFs(
+
SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
filePaths,
new
LastFailingCheckpointStateOutputStreamFactory(
checkpointStreamFactory,
@@ -179,6 +183,7 @@ class ForStStateDataTransferTest extends TestLogger {
assertThatThrownBy(
() ->
stateTransfer.transferFilesToCheckpointFs(
+
SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
Collections.singletonList(first),
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
@@ -222,6 +227,7 @@ class ForStStateDataTransferTest extends TestLogger {
try (ForStStateDataTransfer stateTransfer = new
ForStStateDataTransfer(5)) {
HandleAndLocalPath handleAndLocalPath =
stateTransfer.transferFileToCheckpointFs(
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
sstFile,
headBytes,
checkpointStreamFactory,
@@ -264,6 +270,7 @@ class ForStStateDataTransferTest extends TestLogger {
try (ForStStateDataTransfer stateTransfer = new
ForStStateDataTransfer(5)) {
List<HandleAndLocalPath> sstFiles =
stateTransfer.transferFilesToCheckpointFs(
+ SnapshotType.SharingFilesStrategy.FORWARD_BACKWARD,
sstFilePaths,
checkpointStreamFactory,
CheckpointedStateScope.SHARED,
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/FileMappingManagerTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/FileMappingManagerTest.java
index bf3d1df7b8a..ff7875be8cb 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/FileMappingManagerTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/FileMappingManagerTest.java
@@ -18,13 +18,11 @@
package org.apache.flink.state.forst.fs;
-import org.apache.flink.core.execution.RecoveryClaimMode;
import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.state.forst.fs.filemapping.FileMappingManager;
-import org.apache.flink.state.forst.fs.filemapping.FileOwnershipDecider;
import org.apache.flink.state.forst.fs.filemapping.MappingEntry;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
@@ -52,10 +50,6 @@ public class FileMappingManagerTest {
@Parameter public boolean reuseCp;
- FileOwnershipDecider createFileOwnershipDecider() {
- return new FileOwnershipDecider(RecoveryClaimMode.CLAIM);
- }
-
private MappingEntry registerFile(FileMappingManager manager, Path
filePath) {
if (reuseCp) {
return manager.registerReusedRestoredFile(
@@ -69,11 +63,7 @@ public class FileMappingManagerTest {
void testFileLink() throws IOException {
FileSystem localFS = FileSystem.getLocalFileSystem();
FileMappingManager fileMappingManager =
- new FileMappingManager(
- localFS,
- createFileOwnershipDecider(),
- tempDir.toString(),
- tempDir.toString());
+ new FileMappingManager(localFS, tempDir.toString(),
tempDir.toString());
String src = tempDir + "/source";
FSDataOutputStream os = localFS.create(new Path(src),
FileSystem.WriteMode.OVERWRITE);
os.write(233);
@@ -92,11 +82,7 @@ public class FileMappingManagerTest {
// link d->c
FileSystem localFS = FileSystem.getLocalFileSystem();
FileMappingManager fileMappingManager =
- new FileMappingManager(
- localFS,
- createFileOwnershipDecider(),
- tempDir.toString(),
- tempDir.toString());
+ new FileMappingManager(localFS, tempDir.toString(),
tempDir.toString());
String src = tempDir + "/a";
FSDataOutputStream os = localFS.create(new Path(src),
FileSystem.WriteMode.OVERWRITE);
os.write(233);
@@ -127,11 +113,7 @@ public class FileMappingManagerTest {
void testFileDelete() throws IOException {
FileSystem localFS = FileSystem.getLocalFileSystem();
FileMappingManager fileMappingManager =
- new FileMappingManager(
- localFS,
- createFileOwnershipDecider(),
- tempDir.toString(),
- tempDir.toString());
+ new FileMappingManager(localFS, tempDir.toString(),
tempDir.toString());
String src = tempDir + "/source";
registerFile(fileMappingManager, new Path(src));
Path srcFileRealPath =
fileMappingManager.mappingEntry(src).getSourcePath();
@@ -160,10 +142,7 @@ public class FileMappingManagerTest {
FileSystem localFS = FileSystem.getLocalFileSystem();
FileMappingManager fileMappingManager =
new FileMappingManager(
- localFS,
- createFileOwnershipDecider(),
- tempDir.toString() + "/db",
- tempDir.toString() + "/db");
+ localFS, tempDir.toString() + "/db",
tempDir.toString() + "/db");
String testDir = tempDir + "/testDir";
localFS.mkdirs(new Path(testDir));
String src = testDir + "/source";
@@ -190,10 +169,7 @@ public class FileMappingManagerTest {
FileSystem localFS = FileSystem.getLocalFileSystem();
FileMappingManager fileMappingManager =
new FileMappingManager(
- localFS,
- createFileOwnershipDecider(),
- tempDir.toString() + "/db",
- tempDir.toString() + "/db");
+ localFS, tempDir.toString() + "/db",
tempDir.toString() + "/db");
String testDir = tempDir + "/testDir";
localFS.mkdirs(new Path(testDir));
String src = testDir + "/source";
@@ -245,10 +221,7 @@ public class FileMappingManagerTest {
FileSystem localFS = FileSystem.getLocalFileSystem();
FileMappingManager fileMappingManager =
new FileMappingManager(
- localFS,
- createFileOwnershipDecider(),
- tempDir.toString() + "/db",
- tempDir.toString() + "/db");
+ localFS, tempDir.toString() + "/db",
tempDir.toString() + "/db");
String testDir = tempDir + "/testDir";
localFS.mkdirs(new Path(testDir));
String src = testDir + "/source";
diff --git
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java
index 722837727b6..406604c3497 100644
---
a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java
+++
b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystemTest.java
@@ -86,7 +86,6 @@ public class ForStFlinkFileSystemTest {
ForStFlinkFileSystem.get(
new URI(tempDir.toString()),
new org.apache.flink.core.fs.Path(tempDir.toString()),
- null,
null);
testReadAndWriteWithByteBuffer(fileSystem);
}
@@ -98,7 +97,6 @@ public class ForStFlinkFileSystemTest {
new ByteBufferReadableLocalFileSystem(),
tempDir.toString(),
tempDir.toString(),
- null,
fileBasedCache);
testReadAndWriteWithByteBuffer(fileSystem);
}
@@ -178,7 +176,6 @@ public class ForStFlinkFileSystemTest {
new ByteBufferReadableLocalFileSystem(),
tempDir.toString(),
tempDir.toString(),
- null,
fileBasedCache);
org.apache.flink.core.fs.Path testFilePath =
@@ -205,7 +202,6 @@ public class ForStFlinkFileSystemTest {
new ByteBufferReadableLocalFileSystem(),
remotePath.toString(),
localPath.toString(),
- null,
fileBasedCache);
fileSystem.mkdirs(remotePath);
fileSystem.mkdirs(localPath);
@@ -265,7 +261,6 @@ public class ForStFlinkFileSystemTest {
new ByteBufferReadableLocalFileSystem(),
remotePath.toString(),
localPath.toString(),
- null,
cache);
fileSystem.mkdirs(remotePath);
fileSystem.mkdirs(localPath);
