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 2024/06/18 10:47:44 UTC
(flink) 01/03: [FLINK-32086][checkpointing] Normalize file-merging sub dir
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
commit 6723ad7c9fae25e751e31821567a29a000d89f1b
Author: wangfeifan <zo...@163.com>
AuthorDate: Thu Jun 13 02:09:26 2024 +0800
[FLINK-32086][checkpointing] Normalize file-merging sub dir
---
.../filemerging/FileMergingSnapshotManager.java | 14 ++++++++++++--
.../FileMergingSnapshotManagerBuilder.java | 20 ++++++++++++++------
.../state/TaskExecutorFileMergingManager.java | 4 +++-
.../flink/runtime/taskexecutor/TaskExecutor.java | 1 +
.../FileMergingSnapshotManagerTestBase.java | 10 +++++-----
.../runtime/state/OperatorStateBackendTest.java | 9 +++++++--
.../flink/runtime/state/SharedStateRegistryTest.java | 13 +++++++++----
.../state/TaskExecutorFileMergingManagerTest.java | 10 ++++++++++
.../FsMergingCheckpointStorageLocationTest.java | 13 +++++++++++--
9 files changed, 72 insertions(+), 22 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManager.java
index 4a101be5674..f7bd0c7c4ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManager.java
@@ -204,6 +204,8 @@ public interface FileMergingSnapshotManager extends Closeable {
* the parallelism. Note that this key should be consistent across job attempts.
*/
final class SubtaskKey {
+ private static final String MANAGED_DIR_FORMAT = "job_%s_op_%s_%d_%d";
+
final String jobIDString;
final String operatorIDString;
final int subtaskIndex;
@@ -244,6 +246,11 @@ public interface FileMergingSnapshotManager extends Closeable {
environment.getTaskInfo());
}
+ @VisibleForTesting
+ public String getJobIDString() {
+ return jobIDString;
+ }
+
/**
* Generate an unique managed directory name for one subtask.
*
@@ -251,8 +258,11 @@ public interface FileMergingSnapshotManager extends Closeable {
*/
public String getManagedDirName() {
return String.format(
- "%s_%s_%d_%d_",
- jobIDString, operatorIDString, subtaskIndex, parallelism)
+ MANAGED_DIR_FORMAT,
+ jobIDString,
+ operatorIDString,
+ subtaskIndex,
+ parallelism)
.replaceAll("[^a-zA-Z0-9\\-]", "_");
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBuilder.java
index 2f62acded2f..6bf8b2ca0a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBuilder.java
@@ -19,6 +19,8 @@ package org.apache.flink.runtime.checkpoint.filemerging;
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -28,8 +30,12 @@ import java.util.concurrent.Executor;
/** A builder that builds the {@link FileMergingSnapshotManager}. */
public class FileMergingSnapshotManagerBuilder {
- /** The id for identifying a {@link FileMergingSnapshotManager}. */
- private final String id;
+ // Id format for FileMergingSnapshotManager, consist with jobId and tmId
+ private static final String ID_FORMAT = "job_%s_tm_%s";
+
+ private final JobID jobId;
+
+ private final ResourceID tmResourceId;
/** The file merging type. */
private final FileMergingType fileMergingType;
@@ -52,8 +58,10 @@ public class FileMergingSnapshotManagerBuilder {
*
* @param id the id of the manager.
*/
- public FileMergingSnapshotManagerBuilder(String id, FileMergingType type) {
- this.id = id;
+ public FileMergingSnapshotManagerBuilder(
+ JobID jobId, ResourceID tmResourceId, FileMergingType type) {
+ this.jobId = jobId;
+ this.tmResourceId = tmResourceId;
this.fileMergingType = type;
}
@@ -104,7 +112,7 @@ public class FileMergingSnapshotManagerBuilder {
switch (fileMergingType) {
case MERGE_WITHIN_CHECKPOINT:
return new WithinCheckpointFileMergingSnapshotManager(
- id,
+ String.format(ID_FORMAT, jobId, tmResourceId),
maxFileSize,
filePoolType,
maxSpaceAmplification,
@@ -115,7 +123,7 @@ public class FileMergingSnapshotManagerBuilder {
: metricGroup);
case MERGE_ACROSS_CHECKPOINT:
return new AcrossCheckpointFileMergingSnapshotManager(
- id,
+ String.format(ID_FORMAT, jobId, tmResourceId),
maxFileSize,
filePoolType,
maxSpaceAmplification,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManager.java
index e7325ba2f0b..c414655723e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManager.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManage
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBuilder;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingType;
import org.apache.flink.runtime.checkpoint.filemerging.PhysicalFilePool;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
import org.apache.flink.util.ShutdownHookUtil;
@@ -83,6 +84,7 @@ public class TaskExecutorFileMergingManager {
*/
public @Nullable FileMergingSnapshotManager fileMergingSnapshotManagerForTask(
@Nonnull JobID jobId,
+ @Nonnull ResourceID tmResourceId,
@Nonnull ExecutionAttemptID executionAttemptID,
Configuration clusterConfiguration,
Configuration jobConfiguration,
@@ -131,7 +133,7 @@ public class TaskExecutorFileMergingManager {
fileMergingSnapshotManagerAndRetainedExecutions =
Tuple2.of(
new FileMergingSnapshotManagerBuilder(
- jobId.toString(), fileMergingType)
+ jobId, tmResourceId, fileMergingType)
.setMaxFileSize(maxFileSize.getBytes())
.setFilePoolType(
usingBlockingPool
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index 1a008a4be18..f90f0dce4ba 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -779,6 +779,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
final FileMergingSnapshotManager fileMergingSnapshotManager =
fileMergingManager.fileMergingSnapshotManagerForTask(
jobId,
+ getResourceID(),
tdd.getExecutionAttemptId(),
taskManagerConfiguration.getConfiguration(),
jobInformation.getJobConfiguration(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java
index 6a601bd708b..19d18b8c72b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SpaceStat;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SubtaskKey;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.state.CheckpointedStateScope;
@@ -83,13 +84,11 @@ public abstract class FileMergingSnapshotManagerTestBase {
@BeforeEach
public void setup(@TempDir java.nio.file.Path tempFolder) {
- // use simplified job ids for the tests
- long jobId = 1;
subtaskKey1 =
new SubtaskKey(jobID, operatorID, new TaskInfoImpl("TestingTask", 128, 0, 128, 3));
subtaskKey2 =
new SubtaskKey(jobID, operatorID, new TaskInfoImpl("TestingTask", 128, 1, 128, 3));
- checkpointBaseDir = new Path(tempFolder.toString(), String.valueOf(jobId));
+ checkpointBaseDir = new Path(tempFolder.toString(), jobID.toHexString());
writeBufferSize = 4096;
}
@@ -106,7 +105,7 @@ public abstract class FileMergingSnapshotManagerTestBase {
AbstractFsCheckpointStorageAccess
.CHECKPOINT_TASK_OWNED_STATE_DIR
+ "/"
- + tmId));
+ + String.format("job_%s_tm_%s", jobID, tmId)));
assertThat(fmsm.getManagedDir(subtaskKey1, CheckpointedStateScope.SHARED))
.isEqualTo(
new Path(
@@ -591,7 +590,8 @@ public abstract class FileMergingSnapshotManagerTestBase {
fs.mkdirs(taskOwnedStateDir);
}
FileMergingSnapshotManager fmsm =
- new FileMergingSnapshotManagerBuilder(tmId, getFileMergingType())
+ new FileMergingSnapshotManagerBuilder(
+ jobID, new ResourceID(tmId), getFileMergingType())
.setMaxFileSize(maxFileSize)
.setFilePoolType(filePoolType)
.setMaxSpaceAmplification(spaceAmplification)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
index ace6bffa9c0..3c7a2f4f3f0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.state;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.serialization.SerializerConfigImpl;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ListState;
@@ -41,6 +42,7 @@ import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManage
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SubtaskKey;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBuilder;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingType;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.state.filemerging.FileMergingOperatorStreamStateHandle;
import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess;
@@ -451,8 +453,9 @@ class OperatorStateBackendTest {
checkpointBaseDir,
AbstractFsCheckpointStorageAccess.CHECKPOINT_TASK_OWNED_STATE_DIR);
+ JobID jobId = JobID.generate();
final FileMergingSnapshotManager.SubtaskKey subtaskKey =
- new FileMergingSnapshotManager.SubtaskKey("jobId", "opId", 1, 1);
+ new FileMergingSnapshotManager.SubtaskKey(jobId.toHexString(), "opId", 1, 1);
LocalFileSystem fs = getSharedInstance();
CheckpointStorageLocationReference cslReference =
AbstractFsCheckpointStorageAccess.encodePathAsReference(
@@ -1123,7 +1126,9 @@ class OperatorStateBackendTest {
SubtaskKey subtaskKey) {
FileMergingSnapshotManager mgr =
new FileMergingSnapshotManagerBuilder(
- "test-1", FileMergingType.MERGE_WITHIN_CHECKPOINT)
+ JobID.fromHexString(subtaskKey.getJobIDString()),
+ new ResourceID("test-1"),
+ FileMergingType.MERGE_WITHIN_CHECKPOINT)
.build();
mgr.initFileSystem(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java
index 7e6a2e14763..39d8779c1d4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SharedStateRegistryTest.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBuilder;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingType;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle.ChangelogStateBackendHandleImpl;
import org.apache.flink.runtime.state.filemerging.EmptyFileMergingOperatorStreamStateHandle;
@@ -270,6 +271,8 @@ class SharedStateRegistryTest {
void testFireMergingOperatorStateRegister(@TempDir File tmpFolder) throws IOException {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
+ JobID jobId = JobID.generate();
+
Path checkpointBaseDir = new Path(tmpFolder.toString());
Path sharedStateDir =
new Path(
@@ -280,10 +283,10 @@ class SharedStateRegistryTest {
checkpointBaseDir,
AbstractFsCheckpointStorageAccess.CHECKPOINT_TASK_OWNED_STATE_DIR);
final FileMergingSnapshotManager.SubtaskKey subtaskKey =
- new FileMergingSnapshotManager.SubtaskKey("jobId", "opId", 1, 2);
+ new FileMergingSnapshotManager.SubtaskKey(jobId.toHexString(), "opId", 1, 2);
FileMergingSnapshotManager snapshotManager =
createFileMergingSnapshotManager(
- checkpointBaseDir, sharedStateDir, taskOwnedStateDir);
+ jobId, checkpointBaseDir, sharedStateDir, taskOwnedStateDir);
snapshotManager.registerSubtaskForSharedStates(subtaskKey);
FileMergingOperatorStreamStateHandle handle1 =
EmptyFileMergingOperatorStreamStateHandle.create(
@@ -330,10 +333,12 @@ class SharedStateRegistryTest {
}
private FileMergingSnapshotManager createFileMergingSnapshotManager(
- Path checkpointBaseDir, Path sharedStateDir, Path taskOwnedStateDir) {
+ JobID jobId, Path checkpointBaseDir, Path sharedStateDir, Path taskOwnedStateDir) {
FileMergingSnapshotManager mgr =
new FileMergingSnapshotManagerBuilder(
- "test-1", FileMergingType.MERGE_WITHIN_CHECKPOINT)
+ jobId,
+ new ResourceID("test-1"),
+ FileMergingType.MERGE_WITHIN_CHECKPOINT)
.build();
mgr.initFileSystem(
getSharedInstance(), checkpointBaseDir, sharedStateDir, taskOwnedStateDir, 1024);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManagerTest.java
index cc0d3f16696..de59d1cea4f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorFileMergingManagerTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SubtaskKey;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
@@ -37,6 +38,11 @@ import static org.assertj.core.api.Assertions.assertThat;
public class TaskExecutorFileMergingManagerTest {
@Test
public void testCheckpointScope(@TempDir java.nio.file.Path testBaseDir) throws IOException {
+ ResourceID tmResourceId1 = ResourceID.generate();
+ ResourceID tmResourceId2 = ResourceID.generate();
+ ResourceID tmResourceId3 = ResourceID.generate();
+ ResourceID tmResourceId4 = ResourceID.generate();
+
TaskExecutorFileMergingManager taskExecutorFileMergingManager =
new TaskExecutorFileMergingManager();
JobID job1 = new JobID(1234L, 4321L);
@@ -53,6 +59,7 @@ public class TaskExecutorFileMergingManagerTest {
FileMergingSnapshotManager manager1 =
taskExecutorFileMergingManager.fileMergingSnapshotManagerForTask(
job1,
+ tmResourceId1,
executionID1,
clusterConfig,
jobConfig,
@@ -68,6 +75,7 @@ public class TaskExecutorFileMergingManagerTest {
FileMergingSnapshotManager manager2 =
taskExecutorFileMergingManager.fileMergingSnapshotManagerForTask(
job1,
+ tmResourceId2,
executionID2,
clusterConfig,
jobConfig,
@@ -83,6 +91,7 @@ public class TaskExecutorFileMergingManagerTest {
FileMergingSnapshotManager manager3 =
taskExecutorFileMergingManager.fileMergingSnapshotManagerForTask(
job2,
+ tmResourceId3,
executionID3,
clusterConfig,
jobConfig,
@@ -123,6 +132,7 @@ public class TaskExecutorFileMergingManagerTest {
FileMergingSnapshotManager manager4 =
taskExecutorFileMergingManager.fileMergingSnapshotManagerForTask(
job1,
+ tmResourceId4,
executionID4,
clusterConfig,
jobConfig,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsMergingCheckpointStorageLocationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsMergingCheckpointStorageLocationTest.java
index 858788cd2ac..d70f7fb6ff7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsMergingCheckpointStorageLocationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsMergingCheckpointStorageLocationTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.state.filesystem;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
@@ -25,6 +26,8 @@ import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManagerBuilder;
import org.apache.flink.runtime.checkpoint.filemerging.FileMergingType;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import org.apache.flink.runtime.state.filemerging.SegmentFileStateHandle;
@@ -60,12 +63,16 @@ public class FsMergingCheckpointStorageLocationTest {
private final Random random = new Random();
+ private static final JobID jobId = JobID.generate();
+ private static final OperatorID opId = new OperatorID();
+
private static final String SNAPSHOT_MGR_ID = "snapshotMgrId";
private static final int FILE_STATE_SIZE_THRESHOLD = 1024;
private static final int WRITE_BUFFER_SIZE = 1024;
private static final FileMergingSnapshotManager.SubtaskKey SUBTASK_KEY =
- new FileMergingSnapshotManager.SubtaskKey("jobId", "opId", 1, 1);
+ new FileMergingSnapshotManager.SubtaskKey(
+ jobId.toHexString(), opId.toHexString(), 1, 1);
@Before
public void prepareDirectories() {
@@ -222,7 +229,9 @@ public class FsMergingCheckpointStorageLocationTest {
private FileMergingSnapshotManager createFileMergingSnapshotManager(long maxFileSize) {
FileMergingSnapshotManager mgr =
new FileMergingSnapshotManagerBuilder(
- SNAPSHOT_MGR_ID, FileMergingType.MERGE_WITHIN_CHECKPOINT)
+ jobId,
+ new ResourceID(SNAPSHOT_MGR_ID),
+ FileMergingType.MERGE_WITHIN_CHECKPOINT)
.build();
mgr.initFileSystem(