From fbe2e6bb398f3d000d454e5bb2051cb1507828c7 Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Fri, 3 Jul 2026 19:24:56 +0200 Subject: [PATCH 1/2] [hotfix][tests] Fix SavepointDeepCopyTest checking the source directory twice testSavepointDeepCopy listed savepointPath1 for both stateFiles1 and stateFiles2, so the copied-files assertion compared savepoint1's directory with itself and savepoint2's directory was never examined at all. Point stateFiles2 at savepointPath2, as the variable name, the assertion messages, and the test's documented steps always intended. The mix-up was introduced when the getFileNamesInDirectory helper was extracted (f125067ed0c); before that, both checks listed savepointPath2. Co-Authored-By: Claude Fable 5 --- .../java/org/apache/flink/state/api/SavepointDeepCopyTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java index 144aeb9d1c6dab..a633c59dd05134 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java @@ -173,7 +173,7 @@ public void testSavepointDeepCopy() throws Exception { .write(savepointPath2); env.execute("create savepoint2"); - Set stateFiles2 = getFileNamesInDirectory(Paths.get(savepointPath1)); + Set stateFiles2 = getFileNamesInDirectory(Paths.get(savepointPath2)); Assert.assertTrue( "Failed to create savepoint2 from savepoint1 with additional state files", From 02aeaf4814f897e422ed2d62bf0cd6f3285bf4d0 Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Wed, 8 Jul 2026 17:59:11 +0200 Subject: [PATCH 2/2] [FLINK-39964][state] Fix unrestorable checkpoint after CLAIM restore from native savepoint The first incremental checkpoint after a CLAIM-mode restore from a native savepoint reuses the savepoint's SST files as RelativeFileStateHandles. The metadata serializer stores only their relative path, so on restore they are resolved against the new checkpoint's exclusive directory and can no longer be found. --- .../api/output/SavepointOutputFormat.java | 14 +- .../state/api/SavepointDeepCopyTest.java | 6 + ...avepointOutputFormatSelfContainedTest.java | 175 +++++++ .../flink/runtime/checkpoint/Checkpoints.java | 61 ++- .../metadata/MetadataSerializer.java | 40 +- .../metadata/MetadataV1Serializer.java | 8 +- .../metadata/MetadataV2Serializer.java | 20 +- .../metadata/MetadataV2V3SerializerBase.java | 150 ++++-- .../metadata/MetadataV3Serializer.java | 33 +- .../metadata/MetadataV4Serializer.java | 10 +- .../metadata/MetadataV5Serializer.java | 7 +- .../state/CheckpointMetadataOutputStream.java | 26 +- .../FsCheckpointMetadataOutputStream.java | 5 + .../CheckpointMetadataLoadingTest.java | 2 +- .../runtime/checkpoint/CheckpointsTest.java | 3 +- .../MetadataSerializerEntryPointsTest.java | 185 ++++++++ .../MetadataV2V3SerializerBaseTest.java | 434 ++++++++++++++++++ .../flink/runtime/jobmaster/TestUtils.java | 2 +- .../OperatorCoordinatorSchedulerTest.java | 2 +- ...aimSavepointThenCheckpointRestoreTest.java | 354 ++++++++++++++ ...ointAfterClaimedNativeSavepointITCase.java | 262 +++++++++++ 21 files changed, 1742 insertions(+), 57 deletions(-) create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBaseTest.java create mode 100644 flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBClaimSavepointThenCheckpointRestoreTest.java create mode 100644 flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointAfterClaimedNativeSavepointITCase.java diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java index c0d022e04a78b7..2689479763d5e8 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java @@ -79,7 +79,19 @@ public void writeRecord(CheckpointMetadata metadata) throws IOException { () -> { try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) { - Checkpoints.storeCheckpointMetadata(metadata, out); + // Must stay on the variant WITHOUT the exclusive directory. Files + // retained from an existing savepoint are copied into this + // directory by FileCopyFunction, but their handles still reference + // the source savepoint. Writing without the exclusive directory + // keeps the relative (file-name-only) encoding, which resolves + // against this directory on restore and keeps the savepoint + // self-contained. The exclusive-dir-aware + // Checkpoints.storeCheckpointMetadata would instead fill this + // savepoint's metadata with absolute paths pointing into the + // source savepoint, breaking this savepoint once the source is + // deleted. + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir( + metadata, out); CompletedCheckpointStorageLocation finalizedLocation = out.closeAndFinalizeCheckpoint(); return finalizedLocation.getExternalPointer(); diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java index a633c59dd05134..56f3c6f044f9b1 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java @@ -35,6 +35,7 @@ import org.apache.flink.test.util.AbstractTestBaseJUnit4; import org.apache.flink.util.AbstractID; import org.apache.flink.util.Collector; +import org.apache.flink.util.FileUtils; import org.apache.commons.lang3.RandomStringUtils; import org.junit.Assert; @@ -184,6 +185,11 @@ public void testSavepointDeepCopy() throws Exception { stateFiles1, everyItem(isIn(stateFiles2))); + // Not a cleanup step: deleting savepoint1 before reading savepoint2 proves the deep + // copy is self-contained — savepoint2's metadata must reference the copied files, not + // the originals in savepoint1. + FileUtils.deleteDirectory(new File(savepointPath1)); + // Try to fromExistingSavepoint savepoint2 and read the state of "Operator1" (which has not // been // touched/changed when savepoint2 diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java new file mode 100644 index 00000000000000..08b08758d86b06 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.output; + +import org.apache.flink.api.common.io.FirstAttemptInitializationContext; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +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.RelativeFileStateHandle; +import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; +import org.apache.flink.state.api.runtime.OperatorIDGenerator; +import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.streaming.util.MockStreamingRuntimeContext; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Collections; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that a savepoint written by {@link SavepointOutputFormat} stays self-contained when it + * retains operator state from an existing savepoint, i.e. that it holds no references back into the + * source savepoint it was derived from. + * + *

{@code SavepointWriter.fromExistingSavepoint(...)} keeps retained operator state by copying + * the state files into the new savepoint directory ({@link FileCopyFunction}, keeping the file + * names), while the retained {@link RelativeFileStateHandle}s still carry the paths of the original + * files in the source savepoint. The new savepoint is self-contained only if its metadata + * references those files by file name alone: such a reference resolves against the directory the + * metadata is read from, which is the new savepoint holding the copies. If the metadata recorded + * the handles' absolute paths instead, the new savepoint would keep depending on the source + * savepoint and become unrestorable once the source is deleted. + * + *

See the comment in {@link SavepointOutputFormat#writeRecord} for why the metadata must be + * written with {@code Checkpoints.storeCheckpointMetadataWithoutExclusiveDir}. This test is the + * fast regression guard for that choice; {@code SavepointDeepCopyTest} covers the same guarantee + * end-to-end. + */ +class SavepointOutputFormatSelfContainedTest { + + private static final int PARALLELISM = 1; + private static final int MAX_PARALLELISM = 128; + private static final int SUBTASK_INDEX = 0; + private static final long CHECKPOINT_ID = 1L; + private static final String OPERATOR_UID = "uid"; + private static final String STATE_FILE_NAME = "sst-0001.state"; + + @Test + void testSavepointWithRetainedStateStaysSelfContained(@TempDir File tempDir) throws Exception { + // (1) a source savepoint directory ('sp-old') and the new one being written ('sp-new'). + File sourceSavepointDir = new File(tempDir, "sp-old"); + File targetSavepointDir = new File(tempDir, "sp-new"); + assertThat(sourceSavepointDir.mkdirs()).isTrue(); + assertThat(targetSavepointDir.mkdirs()).isTrue(); + + // (2) a retained state file living in the source savepoint, referenced by a relative + // handle whose full path still points at the file in the source directory (this is what + // loading an existing savepoint produces). + byte[] stateBytes = "retained-shared-state".getBytes(StandardCharsets.UTF_8); + File sourceStateFile = new File(sourceSavepointDir, STATE_FILE_NAME); + Files.write(sourceStateFile.toPath(), stateBytes); + + Path targetSavepointPath = new Path(targetSavepointDir.getAbsolutePath()); + Path sourceStatePath = + new Path(new Path(sourceSavepointDir.getAbsolutePath()), STATE_FILE_NAME); + RelativeFileStateHandle retainedHandle = + new RelativeFileStateHandle(sourceStatePath, STATE_FILE_NAME, stateBytes.length); + + // (3) simulate FileCopyFunction copying the file into the new savepoint directory under + // the same name. + Files.copy( + sourceStateFile.toPath(), new File(targetSavepointDir, STATE_FILE_NAME).toPath()); + + // (4) build metadata that carries the retained relative handle as shared incremental state. + CheckpointMetadata metadata = createMetadata(retainedHandle); + + // (5) write the savepoint metadata into the new directory. + SavepointOutputFormat format = new SavepointOutputFormat(targetSavepointPath); + format.setRuntimeContext(new MockStreamingRuntimeContext(PARALLELISM, SUBTASK_INDEX)); + format.open(FirstAttemptInitializationContext.of(SUBTASK_INDEX, PARALLELISM)); + format.writeRecord(metadata); + format.close(); + + // (6) reload the written metadata, resolving relative handles against the new directory. + CheckpointMetadata reloaded = + SavepointLoader.loadSavepointMetadata(targetSavepointPath.getPath()); + StreamStateHandle reloadedShared = extractSingleSharedStateHandle(reloaded); + + // (7) the reloaded reference must point at the copy inside the new savepoint, not back at + // the original file in the source savepoint. + assertThat(reloadedShared) + .as( + "Retained state must reload as a relative handle that resolves inside " + + "the new savepoint; an absolute path back into the source " + + "savepoint would break the new savepoint once the source is " + + "deleted") + .isInstanceOf(RelativeFileStateHandle.class); + + RelativeFileStateHandle reloadedRelative = (RelativeFileStateHandle) reloadedShared; + assertThat(reloadedRelative.getRelativePath()).isEqualTo(STATE_FILE_NAME); + assertThat(reloadedRelative.getFilePath().getPath()) + .isEqualTo(new Path(targetSavepointPath, STATE_FILE_NAME).getPath()); + assertThat(new File(reloadedRelative.getFilePath().getPath())).exists(); + } + + private static CheckpointMetadata createMetadata(RelativeFileStateHandle retainedHandle) { + StreamStateHandle metaStateHandle = + new ByteStreamStateHandle( + "backend-meta", "backend-meta".getBytes(StandardCharsets.UTF_8)); + IncrementalRemoteKeyedStateHandle keyedStateHandle = + new IncrementalRemoteKeyedStateHandle( + UUID.randomUUID(), + KeyGroupRange.of(0, MAX_PARALLELISM - 1), + CHECKPOINT_ID, + Collections.singletonList( + HandleAndLocalPath.of(retainedHandle, STATE_FILE_NAME)), + Collections.emptyList(), + metaStateHandle); + + OperatorSubtaskState subtaskState = + OperatorSubtaskState.builder().setManagedKeyedState(keyedStateHandle).build(); + + OperatorState operatorState = + new OperatorState( + null, + OPERATOR_UID, + OperatorIDGenerator.fromUid(OPERATOR_UID), + PARALLELISM, + MAX_PARALLELISM); + operatorState.putState(SUBTASK_INDEX, subtaskState); + + return new CheckpointMetadata( + CHECKPOINT_ID, Collections.singleton(operatorState), Collections.emptyList()); + } + + private static StreamStateHandle extractSingleSharedStateHandle(CheckpointMetadata metadata) { + IncrementalRemoteKeyedStateHandle keyedStateHandle = + (IncrementalRemoteKeyedStateHandle) + metadata.getOperatorStates() + .iterator() + .next() + .getState(SUBTASK_INDEX) + .getManagedKeyedState() + .iterator() + .next(); + return keyedStateHandle.getSharedState().get(0).getHandle(); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java index 4f1000fd3b3707..91c856c184dde7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java @@ -20,6 +20,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.OperatorIDPair; import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; import org.apache.flink.runtime.checkpoint.metadata.MetadataSerializer; @@ -28,6 +29,7 @@ import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; import org.apache.flink.runtime.jobgraph.JobVertexID; import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.state.CheckpointMetadataOutputStream; import org.apache.flink.runtime.state.CheckpointStorage; import org.apache.flink.runtime.state.CheckpointStorageLoader; import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation; @@ -76,29 +78,80 @@ public class Checkpoints { // Writing out checkpoint metadata // ------------------------------------------------------------------------ - public static void storeCheckpointMetadata( + /** + * Stores the checkpoint metadata without knowing the checkpoint's exclusive directory: relative + * file references are always persisted as relative and are resolved against whatever directory + * the metadata is later read from. When writing the metadata of an actual checkpoint or + * savepoint, prefer {@link #storeCheckpointMetadata(CheckpointMetadata, + * CheckpointMetadataOutputStream)}. + * + *

The deliberately different method name keeps the context-free variants out of the {@code + * storeCheckpointMetadata} overload set, so a caller cannot switch between the two encodings by + * merely changing a variable's static type. + */ + public static void storeCheckpointMetadataWithoutExclusiveDir( CheckpointMetadata checkpointMetadata, OutputStream out) throws IOException { DataOutputStream dos = new DataOutputStream(out); - storeCheckpointMetadata(checkpointMetadata, dos); + storeCheckpointMetadataWithoutExclusiveDir(checkpointMetadata, dos); } + /** + * Stores the checkpoint metadata into the given metadata output stream, passing the stream's + * exclusive checkpoint directory to the serializer so that relative file references stay + * self-consistent on recovery. + */ public static void storeCheckpointMetadata( + CheckpointMetadata checkpointMetadata, CheckpointMetadataOutputStream out) + throws IOException { + + DataOutputStream dos = new DataOutputStream(out); + storeCheckpointMetadata( + checkpointMetadata, + dos, + MetadataV6Serializer.INSTANCE, + out.getExclusiveCheckpointDir()); + } + + /** + * Variant of {@link #storeCheckpointMetadataWithoutExclusiveDir(CheckpointMetadata, + * OutputStream)} for a {@link DataOutputStream}. + */ + public static void storeCheckpointMetadataWithoutExclusiveDir( CheckpointMetadata checkpointMetadata, DataOutputStream out) throws IOException { - storeCheckpointMetadata(checkpointMetadata, out, MetadataV6Serializer.INSTANCE); + storeCheckpointMetadataWithoutExclusiveDir( + checkpointMetadata, out, MetadataV6Serializer.INSTANCE); } public static void storeCheckpointMetadata( + CheckpointMetadata checkpointMetadata, + DataOutputStream out, + @Nullable Path exclusiveDir) + throws IOException { + storeCheckpointMetadata( + checkpointMetadata, out, MetadataV6Serializer.INSTANCE, exclusiveDir); + } + + public static void storeCheckpointMetadataWithoutExclusiveDir( CheckpointMetadata checkpointMetadata, DataOutputStream out, MetadataSerializer serializer) throws IOException { + storeCheckpointMetadata(checkpointMetadata, out, serializer, null); + } + + public static void storeCheckpointMetadata( + CheckpointMetadata checkpointMetadata, + DataOutputStream out, + MetadataSerializer serializer, + @Nullable Path exclusiveDir) + throws IOException { // write generic header out.writeInt(HEADER_MAGIC_NUMBER); out.writeInt(serializer.getVersion()); - serializer.serialize(checkpointMetadata, out); + serializer.serialize(checkpointMetadata, out, exclusiveDir); } // ------------------------------------------------------------------------ diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java index 543b2c5fc5b07c..04c551e8db0e4c 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java @@ -18,8 +18,11 @@ package org.apache.flink.runtime.checkpoint.metadata; +import org.apache.flink.core.fs.Path; import org.apache.flink.core.io.Versioned; +import javax.annotation.Nullable; + import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; @@ -46,9 +49,42 @@ CheckpointMetadata deserialize( throws IOException; /** - * Serializes a savepoint or checkpoint metadata to an output stream. + * Serializes a savepoint or checkpoint metadata to an output stream without knowledge of the + * exclusive directory the metadata is written into. + * + *

This is a convenience variant of {@link #serialize(CheckpointMetadata, DataOutputStream, + * Path)} with a {@code null} exclusive directory: relative file references keep the relative + * encoding unconditionally and are resolved against whatever directory the metadata is later + * read from, which is only correct if every referenced file actually lives in that directory. + * + * @throws IOException Serialization failures are forwarded + */ + default void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + throws IOException { + serialize(checkpointMetadata, dos, null); + } + + /** + * Serializes a savepoint or checkpoint metadata to an output stream, knowing the exclusive + * directory into which the metadata (and the checkpoint/savepoint it describes) is being + * written. + * + *

The exclusive directory is used to keep the on-disk references self-consistent: a relative + * file handle that does not belong to {@code exclusiveDirPath} (e.g. a savepoint SST reused by + * the first incremental checkpoint after a CLAIM-mode restore) is persisted with its absolute + * path instead of being (incorrectly) resolved against {@code exclusiveDirPath} on recovery. * + *

This is the method implementations must provide (rather than a default dropping the + * exclusive directory), so that a serializer cannot silently lose the exclusive-directory + * information. + * + * @param exclusiveDirPath the exclusive directory of the checkpoint/savepoint being written, or + * {@code null} if it is unknown. * @throws IOException Serialization failures are forwarded */ - void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) throws IOException; + void serialize( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + @Nullable Path exclusiveDirPath) + throws IOException; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV1Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV1Serializer.java index d1e0075f7ae0fc..8d35eb72c4d5bd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV1Serializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV1Serializer.java @@ -19,6 +19,9 @@ package org.apache.flink.runtime.checkpoint.metadata; import org.apache.flink.annotation.Internal; +import org.apache.flink.core.fs.Path; + +import javax.annotation.Nullable; import java.io.DataInputStream; import java.io.DataOutputStream; @@ -52,7 +55,10 @@ public CheckpointMetadata deserialize( } @Override - public void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + public void serialize( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + @Nullable Path exclusiveDirPath) throws IOException { throw new UnsupportedOperationException( "Serialization in v" + getVersion() + " is no longer supported"); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2Serializer.java index c993acb1c4d4a1..8f324864167db3 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2Serializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2Serializer.java @@ -19,6 +19,7 @@ package org.apache.flink.runtime.checkpoint.metadata; import org.apache.flink.annotation.Internal; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.checkpoint.OperatorState; import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; import org.apache.flink.runtime.jobgraph.OperatorID; @@ -67,7 +68,10 @@ public CheckpointMetadata deserialize( } @Override - public void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + public void serialize( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + @Nullable Path exclusiveDirPath) throws IOException { throw new UnsupportedOperationException( "Serialization in v" + getVersion() + " is no longer supported"); @@ -78,7 +82,10 @@ public void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream do // ------------------------------------------------------------------------ @Override - protected void serializeOperatorState(OperatorState operatorState, DataOutputStream dos) + protected void serializeOperatorState( + OperatorState operatorState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { checkState( !operatorState.isFullyFinished(), @@ -102,7 +109,7 @@ protected void serializeOperatorState(OperatorState operatorState, DataOutputStr dos.writeInt(subtaskStateMap.size()); for (Map.Entry entry : subtaskStateMap.entrySet()) { dos.writeInt(entry.getKey()); - serializeSubtaskState(entry.getValue(), dos); + serializeSubtaskState(entry.getValue(), dos, context); } } @@ -134,7 +141,10 @@ protected OperatorState deserializeOperatorState( } @Override - protected void serializeSubtaskState(OperatorSubtaskState subtaskState, DataOutputStream dos) + protected void serializeSubtaskState( + OperatorSubtaskState subtaskState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { // write two unused fields for compatibility: // - "duration" @@ -142,7 +152,7 @@ protected void serializeSubtaskState(OperatorSubtaskState subtaskState, DataOutp dos.writeLong(-1); dos.writeInt(0); - super.serializeSubtaskState(subtaskState, dos); + super.serializeSubtaskState(subtaskState, dos, context); } @Override diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java index 57d69b29f3e561..0c589d0cbd16c0 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java @@ -145,7 +145,10 @@ public abstract class MetadataV2V3SerializerBase { // (De)serialization entry points // ------------------------------------------------------------------------ - protected void serializeMetadata(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + protected void serializeMetadata( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { // first: checkpoint ID dos.writeLong(checkpointMetadata.getCheckpointId()); @@ -162,7 +165,7 @@ protected void serializeMetadata(CheckpointMetadata checkpointMetadata, DataOutp dos.writeInt(operatorStates.size()); for (OperatorState operatorState : operatorStates) { - serializeOperatorState(operatorState, dos); + serializeOperatorState(operatorState, dos, context); } } @@ -267,25 +270,37 @@ protected MasterState deserializeMasterState(DataInputStream dis) throws IOExcep // ------------------------------------------------------------------------ protected abstract void serializeOperatorState( - OperatorState operatorState, DataOutputStream dos) throws IOException; + OperatorState operatorState, + DataOutputStream dos, + @Nullable SerializationContext context) + throws IOException; protected abstract OperatorState deserializeOperatorState( DataInputStream dis, @Nullable DeserializationContext context) throws IOException; - protected void serializeSubtaskState(OperatorSubtaskState subtaskState, DataOutputStream dos) + protected void serializeSubtaskState( + OperatorSubtaskState subtaskState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { serializeSingleton( - subtaskState.getManagedOperatorState(), dos, this::serializeOperatorStateHandle); + subtaskState.getManagedOperatorState(), + dos, + (handle, out) -> serializeOperatorStateHandle(handle, out, context)); serializeSingleton( - subtaskState.getRawOperatorState(), dos, this::serializeOperatorStateHandle); - serializeKeyedStateCol(subtaskState.getManagedKeyedState(), dos); - serializeKeyedStateCol(subtaskState.getRawKeyedState(), dos); + subtaskState.getRawOperatorState(), + dos, + (handle, out) -> serializeOperatorStateHandle(handle, out, context)); + serializeKeyedStateCol(subtaskState.getManagedKeyedState(), dos, context); + serializeKeyedStateCol(subtaskState.getRawKeyedState(), dos, context); } private void serializeKeyedStateCol( - StateObjectCollection managedKeyedState, DataOutputStream dos) + StateObjectCollection managedKeyedState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { - serializeKeyedStateHandle(extractSingleton(managedKeyedState), dos); + serializeKeyedStateHandle(extractSingleton(managedKeyedState), dos, context); } protected OperatorSubtaskState deserializeSubtaskState( @@ -325,6 +340,14 @@ protected OperatorSubtaskState deserializeSubtaskState( @VisibleForTesting static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputStream dos) throws IOException { + serializeKeyedStateHandle(stateHandle, dos, null); + } + + static void serializeKeyedStateHandle( + KeyedStateHandle stateHandle, + DataOutputStream dos, + @Nullable SerializationContext context) + throws IOException { if (stateHandle == null) { dos.writeByte(NULL_HANDLE); } else if (stateHandle instanceof KeyGroupsStateHandle) { @@ -340,7 +363,7 @@ static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputSt for (int keyGroup : keyGroupsStateHandle.getKeyGroupRange()) { dos.writeLong(keyGroupsStateHandle.getOffsetForKeyGroup(keyGroup)); } - serializeStreamStateHandle(keyGroupsStateHandle.getDelegateStateHandle(), dos); + serializeStreamStateHandle(keyGroupsStateHandle.getDelegateStateHandle(), dos, context); // savepoint state handle would not need to persist state handle id out. if (!(stateHandle instanceof KeyGroupsSavepointStateHandle)) { @@ -358,10 +381,13 @@ static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputSt dos.writeInt(incrementalKeyedStateHandle.getKeyGroupRange().getNumberOfKeyGroups()); dos.writeLong(incrementalKeyedStateHandle.getCheckpointedSize()); - serializeStreamStateHandle(incrementalKeyedStateHandle.getMetaDataStateHandle(), dos); + serializeStreamStateHandle( + incrementalKeyedStateHandle.getMetaDataStateHandle(), dos, context); - serializeHandleAndLocalPathList(incrementalKeyedStateHandle.getSharedState(), dos); - serializeHandleAndLocalPathList(incrementalKeyedStateHandle.getPrivateState(), dos); + serializeHandleAndLocalPathList( + incrementalKeyedStateHandle.getSharedState(), dos, context); + serializeHandleAndLocalPathList( + incrementalKeyedStateHandle.getPrivateState(), dos, context); writeStateHandleId(incrementalKeyedStateHandle, dos); } else if (stateHandle instanceof ChangelogStateBackendHandle) { @@ -375,12 +401,12 @@ static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputSt dos.writeInt(handle.getMaterializedStateHandles().size()); for (KeyedStateHandle keyedStateHandle : handle.getMaterializedStateHandles()) { - serializeKeyedStateHandle(keyedStateHandle, dos); + serializeKeyedStateHandle(keyedStateHandle, dos, context); } dos.writeInt(handle.getNonMaterializedStateHandles().size()); for (KeyedStateHandle k : handle.getNonMaterializedStateHandles()) { - serializeKeyedStateHandle(k, dos); + serializeKeyedStateHandle(k, dos, context); } dos.writeLong(handle.getMaterializationID()); @@ -410,7 +436,7 @@ static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputSt for (Tuple2 streamHandleAndOffset : handle.getHandlesAndOffsets()) { dos.writeLong(streamHandleAndOffset.f1); - serializeStreamStateHandle(streamHandleAndOffset.f0, dos); + serializeStreamStateHandle(streamHandleAndOffset.f0, dos, context); } dos.writeLong(handle.getStateSize()); dos.writeLong(handle.getCheckpointedSize()); @@ -596,7 +622,10 @@ private static IncrementalRemoteKeyedStateHandle deserializeIncrementalStateHand stateHandleId); } - void serializeOperatorStateHandle(OperatorStateHandle stateHandle, DataOutputStream dos) + void serializeOperatorStateHandle( + OperatorStateHandle stateHandle, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { if (stateHandle != null) { dos.writeByte( @@ -634,7 +663,7 @@ void serializeOperatorStateHandle(OperatorStateHandle stateHandle, DataOutputStr .toString()); dos.writeBoolean(stateHandle instanceof EmptyFileMergingOperatorStreamStateHandle); } - serializeStreamStateHandle(stateHandle.getDelegateStateHandle(), dos); + serializeStreamStateHandle(stateHandle.getDelegateStateHandle(), dos, context); } else { dos.writeByte(NULL_HANDLE); } @@ -718,12 +747,25 @@ protected void serializeInputStateHandle(InputStateHandle handle, DataOutputStre static void serializeStreamStateHandle(StreamStateHandle stateHandle, DataOutputStream dos) throws IOException { + serializeStreamStateHandle(stateHandle, dos, null); + } + + static void serializeStreamStateHandle( + StreamStateHandle stateHandle, + DataOutputStream dos, + @Nullable SerializationContext context) + throws IOException { if (stateHandle == null) { dos.writeByte(NULL_HANDLE); - } else if (stateHandle instanceof RelativeFileStateHandle) { - dos.writeByte(RELATIVE_STREAM_STATE_HANDLE); + } else if (stateHandle instanceof RelativeFileStateHandle + && canKeepRelativeEncoding((RelativeFileStateHandle) stateHandle, context)) { + // A relative handle keeps the compact RELATIVE_STREAM_STATE_HANDLE encoding only if + // its file lives in the exclusive directory being written. Foreign relative handles + // fall through to the FileStateHandle branch below and are persisted with their + // absolute path. RelativeFileStateHandle relativeFileStateHandle = (RelativeFileStateHandle) stateHandle; + dos.writeByte(RELATIVE_STREAM_STATE_HANDLE); dos.writeUTF(relativeFileStateHandle.getRelativePath()); dos.writeLong(relativeFileStateHandle.getStateSize()); } else if (stateHandle instanceof SegmentFileStateHandle) { @@ -760,13 +802,43 @@ static void serializeStreamStateHandle(StreamStateHandle stateHandle, DataOutput for (int keyGroup : keyGroupsStateHandle.getKeyGroupRange()) { dos.writeLong(keyGroupsStateHandle.getOffsetForKeyGroup(keyGroup)); } - serializeStreamStateHandle(keyGroupsStateHandle.getDelegateStateHandle(), dos); + serializeStreamStateHandle(keyGroupsStateHandle.getDelegateStateHandle(), dos, context); } else { throw new IOException( "Unknown implementation of StreamStateHandle: " + stateHandle.getClass()); } } + /** + * Decides whether a {@link RelativeFileStateHandle} can be written with the relative encoding, + * or has to be written with its absolute path like a plain {@link FileStateHandle}. + * + *

The relative encoding stores only the file name. Whoever reads the metadata later rebuilds + * the full path as {@code /} (see {@link + * #resolveRelativeHandle}). This only works for files that actually live in the directory the + * metadata is written to. Files elsewhere (for example a claimed savepoint's SSTs that a later + * incremental checkpoint still references) would be looked up at a path where they do not + * exist, so their handles must keep the absolute path. If the directory being written to is + * unknown ({@code null}), the legacy behavior applies and the relative encoding is kept. + */ + private static boolean canKeepRelativeEncoding( + RelativeFileStateHandle handle, @Nullable SerializationContext context) { + Path exclusiveDirPath = context == null ? null : context.getExclusiveDirPath(); + return exclusiveDirPath == null + || handle.getFilePath() + .equals(resolveRelativeHandle(exclusiveDirPath, handle.getRelativePath())); + } + + /** + * Rebuilds the absolute path of a {@link RelativeFileStateHandle} from the exclusive directory + * it is resolved against and its relative path. Shared by the write-side encoding decision + * ({@link #canKeepRelativeEncoding}) and the read-side reconstruction in {@code + * deserializeStreamStateHandle} so the two can never drift. + */ + private static Path resolveRelativeHandle(Path exclusiveDir, String relativePath) { + return new Path(exclusiveDir, relativePath); + } + @Nullable static StreamStateHandle deserializeStreamStateHandle( DataInputStream dis, @Nullable DeserializationContext context) throws IOException { @@ -791,7 +863,7 @@ static StreamStateHandle deserializeStreamStateHandle( } String relativePath = dis.readUTF(); long size = dis.readLong(); - Path statePath = new Path(context.getExclusiveDirPath(), relativePath); + Path statePath = resolveRelativeHandle(context.getExclusiveDirPath(), relativePath); return new RelativeFileStateHandle(statePath, relativePath, size); } else if (KEY_GROUPS_HANDLE == type) { @@ -882,12 +954,15 @@ static StateObjectCollection deserializeCollection( } private static void serializeHandleAndLocalPathList( - List list, DataOutputStream dos) throws IOException { + List list, + DataOutputStream dos, + @Nullable SerializationContext context) + throws IOException { dos.writeInt(list.size()); for (HandleAndLocalPath handleAndLocalPath : list) { dos.writeUTF(handleAndLocalPath.getLocalPath()); - serializeStreamStateHandle(handleAndLocalPath.getHandle(), dos); + serializeStreamStateHandle(handleAndLocalPath.getHandle(), dos, context); } } @@ -910,6 +985,29 @@ private static List deserializeHandleAndLocalPathList( // internal helper classes // ------------------------------------------------------------------------ + /** + * A context that keeps information needed while serializing checkpoint metadata. It is + * the write-side counterpart of {@link DeserializationContext} and is passed through the + * serialize methods in the same way. + * + *

It carries the exclusive directory of the checkpoint/savepoint whose metadata is being + * written ({@code null} if unknown), which {@link #canKeepRelativeEncoding} uses to decide how + * a {@link RelativeFileStateHandle} is persisted. + */ + protected static final class SerializationContext { + + @Nullable private final Path exclusiveDirPath; + + SerializationContext(@Nullable Path exclusiveDirPath) { + this.exclusiveDirPath = exclusiveDirPath; + } + + @Nullable + Path getExclusiveDirPath() { + return exclusiveDirPath; + } + } + /** * A context that keeps information needed during serialization. This context is passed along by * the methods. In some sense, this replaces the member fields of the class, because the @@ -925,7 +1023,7 @@ private static List deserializeHandleAndLocalPathList( *

This context is currently hardwired to the FileSystem-based State Backends. At the moment, * this works because those are the only ones producing relative file paths handles, which are * in turn the only ones needing this context. In the future, we should refactor this, though, - * and make the DeserializationContext a property of the used checkpoint storage. That makes + * and make the DeserializationContext a property of the used checkpoint storage. */ protected static final class DeserializationContext { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java index 012914ab14da16..c4d7a5be75925c 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java @@ -20,6 +20,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.checkpoint.FinishedOperatorSubtaskState; import org.apache.flink.runtime.checkpoint.FullyFinishedOperatorState; import org.apache.flink.runtime.checkpoint.OperatorState; @@ -78,9 +79,12 @@ public int getVersion() { // ------------------------------------------------------------------------ @Override - public void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + public void serialize( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + @Nullable Path exclusiveDirPath) throws IOException { - serializeMetadata(checkpointMetadata, dos); + serializeMetadata(checkpointMetadata, dos, new SerializationContext(exclusiveDirPath)); } @Override @@ -95,7 +99,10 @@ public CheckpointMetadata deserialize( // ------------------------------------------------------------------------ @Override - protected void serializeOperatorState(OperatorState operatorState, DataOutputStream dos) + protected void serializeOperatorState( + OperatorState operatorState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { // Operator ID dos.writeLong(operatorState.getOperatorID().getLowerPart()); @@ -106,7 +113,7 @@ protected void serializeOperatorState(OperatorState operatorState, DataOutputStr dos.writeInt(operatorState.getMaxParallelism()); // Coordinator state - serializeStreamStateHandle(operatorState.getCoordinatorState(), dos); + serializeStreamStateHandle(operatorState.getCoordinatorState(), dos, context); // Sub task states if (operatorState.isFullyFinished()) { @@ -119,7 +126,7 @@ protected void serializeOperatorState(OperatorState operatorState, DataOutputStr boolean isFinished = entry.getValue().isFinished(); serializeSubtaskIndexAndFinishedState(entry.getKey(), isFinished, dos); if (!isFinished) { - serializeSubtaskState(entry.getValue(), dos); + serializeSubtaskState(entry.getValue(), dos, context); } } } @@ -137,9 +144,17 @@ private void serializeSubtaskIndexAndFinishedState( } @Override - protected void serializeSubtaskState(OperatorSubtaskState subtaskState, DataOutputStream dos) + protected void serializeSubtaskState( + OperatorSubtaskState subtaskState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { - super.serializeSubtaskState(subtaskState, dos); + super.serializeSubtaskState(subtaskState, dos, context); + // Channel state (unaligned checkpoints) is always written fresh into the current + // checkpoint's own storage, so its stream handles never refer to a foreign exclusive + // directory. It therefore does not need the exclusive-directory-aware encoding (see + // MetadataV2V3SerializerBase#canKeepRelativeEncoding) and keeps using the context-free + // serialization path. serializeCollection( subtaskState.getInputChannelState(), dos, this::serializeInputStateHandle); serializeCollection( @@ -270,7 +285,7 @@ public static StreamStateHandle deserializeStreamStateHandle(DataInputStream dis @VisibleForTesting public void serializeOperatorStateHandleUtil( OperatorStateHandle stateHandle, DataOutputStream dos) throws IOException { - serializeOperatorStateHandle(stateHandle, dos); + serializeOperatorStateHandle(stateHandle, dos, null); } @VisibleForTesting @@ -282,7 +297,7 @@ public OperatorStateHandle deserializeOperatorStateHandleUtil(DataInputStream di @VisibleForTesting public void serializeKeyedStateHandleUtil(KeyedStateHandle stateHandle, DataOutputStream dos) throws IOException { - serializeKeyedStateHandle(stateHandle, dos); + serializeKeyedStateHandle(stateHandle, dos, null); } @VisibleForTesting diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4Serializer.java index 88262fa805a47c..f49be1ce902a52 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4Serializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4Serializer.java @@ -18,8 +18,11 @@ package org.apache.flink.runtime.checkpoint.metadata; import org.apache.flink.annotation.Internal; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.checkpoint.CheckpointProperties; +import javax.annotation.Nullable; + import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; @@ -50,9 +53,12 @@ public CheckpointMetadata deserialize( } @Override - public void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + public void serialize( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + @Nullable Path exclusiveDirPath) throws IOException { - super.serialize(checkpointMetadata, dos); + super.serialize(checkpointMetadata, dos, exclusiveDirPath); serializeProperties(checkpointMetadata.getCheckpointProperties(), dos); } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5Serializer.java index ca0d568ccf4b40..8dd17d1b9ef5e0 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5Serializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5Serializer.java @@ -46,7 +46,10 @@ public int getVersion() { } @Override - protected void serializeOperatorState(OperatorState operatorState, DataOutputStream dos) + protected void serializeOperatorState( + OperatorState operatorState, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { if (operatorState.getOperatorName().isPresent() && operatorState.getOperatorName().get().isEmpty()) { @@ -60,7 +63,7 @@ protected void serializeOperatorState(OperatorState operatorState, DataOutputStr // strings in metadata we do conversion here dos.writeUTF(operatorState.getOperatorName().orElse("")); dos.writeUTF(operatorState.getOperatorUid().orElse("")); - super.serializeOperatorState(operatorState, dos); + super.serializeOperatorState(operatorState, dos, context); } @Override diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointMetadataOutputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointMetadataOutputStream.java index 525e21b0b5eda5..d4a4a4170a7fac 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointMetadataOutputStream.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointMetadataOutputStream.java @@ -19,6 +19,9 @@ package org.apache.flink.runtime.state; import org.apache.flink.core.fs.FSDataOutputStream; +import org.apache.flink.core.fs.Path; + +import javax.annotation.Nullable; import java.io.IOException; @@ -26,7 +29,7 @@ * An output stream for checkpoint metadata. * *

This stream is similar to the {@link CheckpointStateOutputStream}, but for metadata files - * rather thancdata files. + * rather than data files. * *

This stream always creates a file, regardless of the amount of data written. */ @@ -41,6 +44,27 @@ public abstract class CheckpointMetadataOutputStream extends FSDataOutputStream public abstract CompletedCheckpointStorageLocation closeAndFinalizeCheckpoint() throws IOException; + /** + * Returns the exclusive directory of the checkpoint/savepoint whose metadata is written through + * this stream, or {@code null} if it is unknown (e.g. for storage that does not lay out state + * in an exclusive directory). + * + *

This is used while serializing the metadata to keep relative file references + * self-consistent (see {@code MetadataV2V3SerializerBase}). + * + *

Implementations whose state files can be referenced through {@code + * RelativeFileStateHandle}s (in particular any storage laying out state in a per-checkpoint + * exclusive directory) must override this method. Returning {@code null} preserves the legacy + * relative encoding, which is correct as long as every relative handle points into the + * directory the metadata is written to; a relative handle pointing elsewhere would be resolved + * against the wrong directory on recovery, making the checkpoint unrestorable. For storage + * without an exclusive directory layout, returning {@code null} is the right choice. + */ + @Nullable + public Path getExclusiveCheckpointDir() { + return null; + } + /** * This method should close the stream, if has not been closed before. If this method actually * closes the stream, it should delete/release the resource behind the stream, such as the file diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java index 4f5c2596ba02fc..ef6a1120b4ce79 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStream.java @@ -93,6 +93,11 @@ public void sync() throws IOException { outputStreamWrapper.getOutput().sync(); } + @Override + public Path getExclusiveCheckpointDir() { + return exclusiveCheckpointDir; + } + // ------------------------------------------------------------------------ // Closing // ------------------------------------------------------------------------ diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java index 212930044fa737..16cf4b3370510b 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointMetadataLoadingTest.java @@ -202,7 +202,7 @@ private static CompletedCheckpointStorageLocation createSavepointWithOperatorSta final StreamStateHandle serializedMetadata; try (ByteArrayOutputStream os = new ByteArrayOutputStream()) { - Checkpoints.storeCheckpointMetadata(savepoint, os); + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir(savepoint, os); serializedMetadata = new ByteStreamStateHandle("checkpoint", os.toByteArray()); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointsTest.java index 834c73ec4b250e..5c73e68acf8eb5 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointsTest.java @@ -40,7 +40,8 @@ void testVersion3Compatibility() throws IOException { try (ByteArrayOutputStream out = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(out)) { - Checkpoints.storeCheckpointMetadata(metadata, dos, MetadataV3Serializer.INSTANCE); + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir( + metadata, dos, MetadataV3Serializer.INSTANCE); try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(out.toByteArray()))) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java new file mode 100644 index 00000000000000..359cf001c0ee18 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.checkpoint.metadata; + +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.CheckpointProperties; +import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.jobgraph.OperatorID; +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.filesystem.RelativeFileStateHandle; +import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; + +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for the two {@link MetadataSerializer#serialize} entry points (the 2-arg convenience form + * and the 3-arg exclusive-directory-aware form) across every registered metadata version. + * + *

The 3-arg form is the contract-bearing method: the 2-arg form is a default that delegates to + * it with a {@code null} exclusive directory. These tests pin down that contract end-to-end: + * + *

    + *
  • for the writable versions (v3, v4, v5, v6) the two entry points produce byte-identical + * output for a {@code null} exclusive directory (proving the 2-arg form's default delegation + * is behavior-preserving); + *
  • for the same versions, a non-null exclusive directory that makes a shared relative handle + * foreign changes the bytes (proving the directory is threaded all the way through to the + * encoder and not silently dropped); + *
  • the deserialize-only versions (v1, v2) reject serialization through both entry points. + *
+ */ +class MetadataSerializerEntryPointsTest { + + /** Directory the shared relative handle resolves against (its "own" savepoint directory). */ + private static final Path SAVEPOINT_DIR = new Path("s3://bucket/savepoints/sp-1"); + + /** + * A different exclusive directory: relative to it the shared handle is foreign and must be + * promoted to an absolute handle instead of being resolved against it. + */ + private static final Path FOREIGN_EXCLUSIVE_DIR = new Path("s3://bucket/checkpoints/chk-5"); + + private static final String RELATIVE_FILE_NAME = "000001.sst"; + + private static final long STATE_SIZE = 4242L; + + @Test + void testBothEntryPointsProduceIdenticalBytesForNullExclusiveDir() throws Exception { + for (MetadataSerializer serializer : writableSerializers()) { + // Serialize the very same metadata object twice so any random state-handle ids are + // identical between the two runs and only the entry point differs. + final CheckpointMetadata metadata = metadataWithForeignSharedState(); + assertThat(serialize(serializer, metadata)) + .as( + "2-arg and 3-arg(null) entry points must be byte-identical for v%d", + serializer.getVersion()) + .isEqualTo(serialize(serializer, metadata, null)); + } + } + + @Test + void testNonNullExclusiveDirReachesEncoderAndChangesForeignHandleBytes() throws Exception { + for (MetadataSerializer serializer : writableSerializers()) { + final CheckpointMetadata metadata = metadataWithForeignSharedState(); + final byte[] withNullDir = serialize(serializer, metadata, null); + final byte[] withForeignDir = serialize(serializer, metadata, FOREIGN_EXCLUSIVE_DIR); + assertThat(withForeignDir) + .as( + "a non-null exclusive dir must reach the encoder and promote the foreign relative handle to absolute for v%d", + serializer.getVersion()) + .isNotEqualTo(withNullDir); + } + } + + @Test + void testDeserializeOnlyVersionsRejectSerializationThroughBothEntryPoints() { + final CheckpointMetadata metadata = metadataWithForeignSharedState(); + for (MetadataSerializer serializer : deserializeOnlySerializers()) { + assertThatThrownBy(() -> serialize(serializer, metadata)) + .as("2-arg serialize must be unsupported for v%d", serializer.getVersion()) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> serialize(serializer, metadata, null)) + .as("3-arg serialize must be unsupported for v%d", serializer.getVersion()) + .isInstanceOf(UnsupportedOperationException.class); + } + } + + // ------------------------------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------------------------------ + + private static List writableSerializers() { + return Arrays.asList( + MetadataV3Serializer.INSTANCE, + MetadataV4Serializer.INSTANCE, + MetadataV5Serializer.INSTANCE, + MetadataV6Serializer.INSTANCE); + } + + private static List deserializeOnlySerializers() { + return Arrays.asList(MetadataV1Serializer.INSTANCE, MetadataV2Serializer.INSTANCE); + } + + private static CheckpointMetadata metadataWithForeignSharedState() { + final RelativeFileStateHandle sharedHandle = + new RelativeFileStateHandle( + new Path(SAVEPOINT_DIR, RELATIVE_FILE_NAME), + RELATIVE_FILE_NAME, + STATE_SIZE); + final IncrementalRemoteKeyedStateHandle keyedStateHandle = + new IncrementalRemoteKeyedStateHandle( + UUID.nameUUIDFromBytes("backend".getBytes(StandardCharsets.UTF_8)), + KeyGroupRange.of(0, 0), + 1L, + Collections.singletonList( + HandleAndLocalPath.of(sharedHandle, RELATIVE_FILE_NAME)), + Collections.emptyList(), + new ByteStreamStateHandle("meta", new byte[] {1, 2, 3})); + final OperatorSubtaskState subtaskState = + OperatorSubtaskState.builder().setManagedKeyedState(keyedStateHandle).build(); + final OperatorState operatorState = new OperatorState(null, null, new OperatorID(), 1, 1); + operatorState.putState(0, subtaskState); + return new CheckpointMetadata( + 1L, + Collections.singletonList(operatorState), + Collections.emptyList(), + CheckpointProperties.forCheckpoint( + CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION)); + } + + private static byte[] serialize(MetadataSerializer serializer, CheckpointMetadata metadata) + throws IOException { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(baos)) { + serializer.serialize(metadata, dos); + } + return baos.toByteArray(); + } + + private static byte[] serialize( + MetadataSerializer serializer, + CheckpointMetadata metadata, + @Nullable Path exclusiveDirPath) + throws IOException { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(baos)) { + serializer.serialize(metadata, dos, exclusiveDirPath); + } + return baos.toByteArray(); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBaseTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBaseTest.java new file mode 100644 index 00000000000000..ffb73bfebbb6f8 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBaseTest.java @@ -0,0 +1,434 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.checkpoint.metadata; + +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.jobgraph.OperatorID; +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.KeyedStateHandle; +import org.apache.flink.runtime.state.StreamStateHandle; +import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle; +import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess; +import org.apache.flink.runtime.state.filesystem.FileStateHandle; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; +import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; +import org.apache.flink.testutils.junit.utils.TempDirUtils; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for the exclusive-directory-aware serialization of {@link RelativeFileStateHandle}s in + * {@link MetadataV2V3SerializerBase}. + * + *

A {@link RelativeFileStateHandle} only stores the file name relative to the exclusive + * directory it was written into. On recovery the absolute path is rebuilt as {@code new + * Path(, relativePath)}. This is only correct as long as the file physically lives in + * that exclusive directory. The serializer therefore keeps a relative handle relative only when it + * belongs to the exclusive directory being written, and otherwise persists its absolute path. + */ +class MetadataV2V3SerializerBaseTest { + + @TempDir private java.nio.file.Path tmp; + + /** + * A shared-state file that lives outside the checkpoint's exclusive directory (e.g. an + * SST file of a NATIVE savepoint reused by the first incremental checkpoint after a CLAIM-mode + * restore) must survive the metadata round-trip with its correct absolute path, not be resolved + * against the checkpoint's exclusive directory. + */ + @Test + void testForeignRelativeSharedStateRoundTripsAsAbsolute() throws Exception { + final File checkpointDir = TempDirUtils.newFolder(tmp, "chk-3"); + final File savepointDir = TempDirUtils.newFolder(tmp, "savepoint"); + // Make checkpointDir resolvable as an exclusive directory on deserialization. The correct + // encoding never consults it (the foreign handle is written as absolute); if a regression + // instead wrote the handle as relative, this keeps the test failing on the assertion below + // instead of on an unrelated IOException about a missing _metadata file. + writeEmptyMetadataFile(checkpointDir); + + final Path exclusiveDirPath = Path.fromLocalFile(checkpointDir); + final Path savepointDirPath = Path.fromLocalFile(savepointDir); + + final String relativeFileName = UUID.randomUUID().toString(); + // The foreign file physically lives in the savepoint directory, not in checkpointDir. + final Path foreignAbsolutePath = new Path(savepointDirPath, relativeFileName); + final long stateSize = 4242L; + final RelativeFileStateHandle foreignHandle = + new RelativeFileStateHandle(foreignAbsolutePath, relativeFileName, stateSize); + + final IncrementalRemoteKeyedStateHandle reloaded = + roundTrip( + incrementalHandleWithSharedState(foreignHandle, "000001.sst"), + exclusiveDirPath, + checkpointDir.getAbsolutePath()); + + final StreamStateHandle reloadedShared = reloaded.getSharedState().get(0).getHandle(); + assertThat(reloadedShared) + .as("a shared handle pointing outside the exclusive dir must come back absolute") + .isInstanceOf(FileStateHandle.class) + .isNotInstanceOf(RelativeFileStateHandle.class); + final FileStateHandle reloadedFile = (FileStateHandle) reloadedShared; + assertThat(reloadedFile.getFilePath()).isEqualTo(foreignAbsolutePath); + assertThat(reloadedFile.getStateSize()).isEqualTo(stateSize); + } + + /** + * A handle that actually belongs to the exclusive directory being written (a savepoint's own + * SST) must stay relative so that the directory can still be relocated: re-reading the metadata + * from a different location resolves the handle against that new location. + */ + @Test + void testOwnRelativeStateRoundTripsAsRelativeAndIsRelocatable() throws Exception { + final File savepointDir = TempDirUtils.newFolder(tmp, "savepoint"); + // resolveCheckpointPointer (used on deserialization to resolve relative handles) requires + // a _metadata file to exist in the pointed-to directory. + writeEmptyMetadataFile(savepointDir); + + final Path savepointDirPath = Path.fromLocalFile(savepointDir); + final String relativeFileName = UUID.randomUUID().toString(); + final Path ownAbsolutePath = new Path(savepointDirPath, relativeFileName); + final long stateSize = 777L; + final RelativeFileStateHandle ownHandle = + new RelativeFileStateHandle(ownAbsolutePath, relativeFileName, stateSize); + + final byte[] metadataBytes = + serialize( + incrementalHandleWithSharedState(ownHandle, "000002.sst"), + savepointDirPath); + + // Re-read from the original location: must stay relative and resolve to the original path. + final IncrementalRemoteKeyedStateHandle reloaded = + deserialize(metadataBytes, savepointDir.getAbsolutePath()); + + final StreamStateHandle reloadedShared = reloaded.getSharedState().get(0).getHandle(); + assertThat(reloadedShared) + .as("a handle belonging to the exclusive dir must round-trip as relative") + .isInstanceOf(RelativeFileStateHandle.class); + assertThat(((RelativeFileStateHandle) reloadedShared).getRelativePath()) + .isEqualTo(relativeFileName); + assertThat(((FileStateHandle) reloadedShared).getFilePath()).isEqualTo(ownAbsolutePath); + + // Relocate the savepoint: re-read the very same bytes from a *different* directory. Because + // the handle stayed relative, it must now resolve against the new location. + final File movedDir = TempDirUtils.newFolder(tmp, "savepoint-moved"); + writeEmptyMetadataFile(movedDir); + final Path movedDirPath = Path.fromLocalFile(movedDir); + + final IncrementalRemoteKeyedStateHandle relocated = + deserialize(metadataBytes, movedDir.getAbsolutePath()); + final StreamStateHandle relocatedShared = relocated.getSharedState().get(0).getHandle(); + assertThat(relocatedShared).isInstanceOf(RelativeFileStateHandle.class); + assertThat(((FileStateHandle) relocatedShared).getFilePath()) + .as( + "a relocated savepoint must resolve its relative handles against the new directory") + .isEqualTo(new Path(movedDirPath, relativeFileName)); + } + + /** + * A checkpoint may reference its own relative handles and foreign ones (reused savepoint SSTs) + * at the same time. Each must be encoded on its own merits: own handles stay relative, foreign + * handles become absolute. + */ + @Test + void testMixedOwnAndForeignRelativeSharedState() throws Exception { + final File checkpointDir = TempDirUtils.newFolder(tmp, "chk-4"); + final File savepointDir = TempDirUtils.newFolder(tmp, "savepoint"); + writeEmptyMetadataFile(checkpointDir); + + final Path exclusiveDirPath = Path.fromLocalFile(checkpointDir); + final Path savepointDirPath = Path.fromLocalFile(savepointDir); + + final String ownFileName = UUID.randomUUID().toString(); + final String foreignFileName = UUID.randomUUID().toString(); + final RelativeFileStateHandle ownHandle = + new RelativeFileStateHandle( + new Path(exclusiveDirPath, ownFileName), ownFileName, 11L); + final RelativeFileStateHandle foreignHandle = + new RelativeFileStateHandle( + new Path(savepointDirPath, foreignFileName), foreignFileName, 22L); + + final IncrementalRemoteKeyedStateHandle handle = + incrementalHandleWithSharedState( + Arrays.asList( + HandleAndLocalPath.of(ownHandle, "000001.sst"), + HandleAndLocalPath.of(foreignHandle, "000002.sst"))); + + final IncrementalRemoteKeyedStateHandle reloaded = + roundTrip(handle, exclusiveDirPath, checkpointDir.getAbsolutePath()); + + final StreamStateHandle reloadedOwn = reloaded.getSharedState().get(0).getHandle(); + assertThat(reloadedOwn).isInstanceOf(RelativeFileStateHandle.class); + assertThat(((FileStateHandle) reloadedOwn).getFilePath()) + .isEqualTo(new Path(exclusiveDirPath, ownFileName)); + + final StreamStateHandle reloadedForeign = reloaded.getSharedState().get(1).getHandle(); + assertThat(reloadedForeign) + .isInstanceOf(FileStateHandle.class) + .isNotInstanceOf(RelativeFileStateHandle.class); + assertThat(((FileStateHandle) reloadedForeign).getFilePath()) + .isEqualTo(new Path(savepointDirPath, foreignFileName)); + } + + /** + * Without a known exclusive directory (the context-free {@link MetadataSerializer#serialize} + * variant, used e.g. by the state processor API), the legacy encoding must be preserved: every + * relative handle stays relative and is resolved against whatever directory the metadata is + * later read from. + */ + @Test + void testNullExclusiveDirKeepsLegacyRelativeEncoding() throws Exception { + final File checkpointDir = TempDirUtils.newFolder(tmp, "chk-5"); + final File savepointDir = TempDirUtils.newFolder(tmp, "savepoint"); + writeEmptyMetadataFile(checkpointDir); + + final String fileName = UUID.randomUUID().toString(); + final RelativeFileStateHandle foreignHandle = + new RelativeFileStateHandle( + new Path(Path.fromLocalFile(savepointDir), fileName), fileName, 33L); + + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(baos)) { + // context-free convenience variant: must behave like exclusiveDirPath == null + MetadataV3Serializer.INSTANCE.serialize( + metadataFor(incrementalHandleWithSharedState(foreignHandle, "000001.sst")), + dos); + } + + final IncrementalRemoteKeyedStateHandle reloaded = + deserialize(baos.toByteArray(), checkpointDir.getAbsolutePath()); + final StreamStateHandle reloadedShared = reloaded.getSharedState().get(0).getHandle(); + assertThat(reloadedShared).isInstanceOf(RelativeFileStateHandle.class); + assertThat(((FileStateHandle) reloadedShared).getFilePath()) + .as("legacy encoding resolves the handle against the directory it is read from") + .isEqualTo(new Path(Path.fromLocalFile(checkpointDir), fileName)); + } + + /** + * The own/foreign decision must also work for object-store URIs (no filesystem access is + * involved in taking it): a foreign relative handle on s3/abfss is persisted with its full + * absolute URI. + */ + @Test + void testForeignRelativeHandleOnObjectStoreIsWrittenAbsolute() throws Exception { + for (String scheme : new String[] {"s3", "abfss"}) { + final Path savepointDir = new Path(scheme + "://bucket/flink/savepoints/savepoint-1"); + final Path exclusiveDir = new Path(scheme + "://bucket/flink/checkpoints/chk-2"); + final String fileName = UUID.randomUUID().toString(); + final RelativeFileStateHandle foreignHandle = + new RelativeFileStateHandle(new Path(savepointDir, fileName), fileName, 44L); + + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(baos)) { + MetadataV2V3SerializerBase.serializeStreamStateHandle( + foreignHandle, + dos, + new MetadataV2V3SerializerBase.SerializationContext(exclusiveDir)); + } + + // a null deserialization context suffices exactly because the handle must have been + // written absolute; a (wrongly) relative handle would fail to deserialize here + final StreamStateHandle reloaded = + MetadataV2V3SerializerBase.deserializeStreamStateHandle( + new DataInputStream(new ByteArrayInputStream(baos.toByteArray())), + null); + assertThat(reloaded) + .as(scheme + " foreign handle must be written absolute") + .isInstanceOf(FileStateHandle.class) + .isNotInstanceOf(RelativeFileStateHandle.class); + assertThat(((FileStateHandle) reloaded).getFilePath()) + .isEqualTo(new Path(savepointDir, fileName)); + } + } + + /** + * Counterpart of {@link #testForeignRelativeHandleOnObjectStoreIsWrittenAbsolute}: a handle + * that belongs to the object-store exclusive directory keeps the relative encoding, bit for bit + * the same as the legacy (context-free) encoding. + */ + @Test + void testOwnRelativeHandleOnObjectStoreStaysRelative() throws Exception { + for (String scheme : new String[] {"s3", "abfss"}) { + final Path exclusiveDir = new Path(scheme + "://bucket/flink/checkpoints/chk-2"); + final String fileName = UUID.randomUUID().toString(); + final RelativeFileStateHandle ownHandle = + new RelativeFileStateHandle(new Path(exclusiveDir, fileName), fileName, 55L); + + final ByteArrayOutputStream withContext = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(withContext)) { + MetadataV2V3SerializerBase.serializeStreamStateHandle( + ownHandle, + dos, + new MetadataV2V3SerializerBase.SerializationContext(exclusiveDir)); + } + final ByteArrayOutputStream legacy = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(legacy)) { + MetadataV2V3SerializerBase.serializeStreamStateHandle(ownHandle, dos, null); + } + + assertThat(withContext.toByteArray()) + .as(scheme + " own handle must keep the (relative) legacy encoding") + .isEqualTo(legacy.toByteArray()); + } + } + + /** + * The own/foreign decision must also apply to handles nested inside a {@link + * ChangelogStateBackendHandle}: its materialized handles are serialized through a recursive + * call that has to keep passing the serialization context along. + */ + @Test + void testForeignRelativeStateInsideChangelogHandleRoundTripsAsAbsolute() throws Exception { + final File checkpointDir = TempDirUtils.newFolder(tmp, "chk-6"); + final File savepointDir = TempDirUtils.newFolder(tmp, "savepoint"); + writeEmptyMetadataFile(checkpointDir); + + final Path exclusiveDirPath = Path.fromLocalFile(checkpointDir); + final String relativeFileName = UUID.randomUUID().toString(); + final Path foreignAbsolutePath = + new Path(Path.fromLocalFile(savepointDir), relativeFileName); + final RelativeFileStateHandle foreignHandle = + new RelativeFileStateHandle(foreignAbsolutePath, relativeFileName, 4242L); + + // Reachable only through serializeKeyedStateHandle's materialized-handles recursion. + final ChangelogStateBackendHandle changelogHandle = + new ChangelogStateBackendHandle.ChangelogStateBackendHandleImpl( + Collections.singletonList( + incrementalHandleWithSharedState(foreignHandle, "000001.sst")), + Collections.emptyList(), + KeyGroupRange.of(0, 0), + 1L, + 1L, + 0L); + + final KeyedStateHandle reloaded = + deserializeManagedKeyedState( + serialize(changelogHandle, exclusiveDirPath), + checkpointDir.getAbsolutePath()); + + assertThat(reloaded).isInstanceOf(ChangelogStateBackendHandle.class); + final StreamStateHandle reloadedShared = + ((IncrementalRemoteKeyedStateHandle) + ((ChangelogStateBackendHandle) reloaded) + .getMaterializedStateHandles() + .get(0)) + .getSharedState() + .get(0) + .getHandle(); + assertThat(reloadedShared) + .as("a foreign shared handle nested in a changelog handle must come back absolute") + .isInstanceOf(FileStateHandle.class) + .isNotInstanceOf(RelativeFileStateHandle.class); + assertThat(((FileStateHandle) reloadedShared).getFilePath()).isEqualTo(foreignAbsolutePath); + } + + // ------------------------------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------------------------------ + + private static IncrementalRemoteKeyedStateHandle incrementalHandleWithSharedState( + StreamStateHandle sharedHandle, String localPath) { + return incrementalHandleWithSharedState( + Collections.singletonList(HandleAndLocalPath.of(sharedHandle, localPath))); + } + + private static IncrementalRemoteKeyedStateHandle incrementalHandleWithSharedState( + List sharedState) { + return new IncrementalRemoteKeyedStateHandle( + UUID.nameUUIDFromBytes("backend".getBytes(StandardCharsets.UTF_8)), + KeyGroupRange.of(0, 0), + 1L, + sharedState, + Collections.emptyList(), + new ByteStreamStateHandle("meta", new byte[] {1, 2, 3})); + } + + private static IncrementalRemoteKeyedStateHandle roundTrip( + IncrementalRemoteKeyedStateHandle handle, Path exclusiveDirPath, String externalPointer) + throws Exception { + return deserialize(serialize(handle, exclusiveDirPath), externalPointer); + } + + private static CheckpointMetadata metadataFor(KeyedStateHandle handle) { + final OperatorSubtaskState subtaskState = + OperatorSubtaskState.builder().setManagedKeyedState(handle).build(); + final OperatorState operatorState = new OperatorState(null, null, new OperatorID(), 1, 1); + operatorState.putState(0, subtaskState); + return new CheckpointMetadata( + 1L, Collections.singletonList(operatorState), Collections.emptyList()); + } + + private static byte[] serialize(KeyedStateHandle handle, Path exclusiveDirPath) + throws Exception { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(baos)) { + MetadataV3Serializer.INSTANCE.serialize(metadataFor(handle), dos, exclusiveDirPath); + } + return baos.toByteArray(); + } + + private static IncrementalRemoteKeyedStateHandle deserialize( + byte[] bytes, String externalPointer) throws Exception { + return (IncrementalRemoteKeyedStateHandle) + deserializeManagedKeyedState(bytes, externalPointer); + } + + private static KeyedStateHandle deserializeManagedKeyedState( + byte[] bytes, String externalPointer) throws Exception { + final CheckpointMetadata reloaded; + try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(bytes))) { + reloaded = + MetadataV3Serializer.INSTANCE.deserialize( + dis, + MetadataV2V3SerializerBaseTest.class.getClassLoader(), + externalPointer); + } + final OperatorState operatorState = reloaded.getOperatorStates().iterator().next(); + final OperatorSubtaskState subtaskState = operatorState.getStates().iterator().next(); + return subtaskState.getManagedKeyedState().iterator().next(); + } + + private static void writeEmptyMetadataFile(File dir) throws Exception { + final Path metadataFile = + new Path( + Path.fromLocalFile(dir), + AbstractFsCheckpointStorageAccess.METADATA_FILE_NAME); + FileSystem.getLocalFileSystem() + .create(metadataFile, FileSystem.WriteMode.OVERWRITE) + .close(); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java index 0b51a13734ccc6..49ca4c54ea18af 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestUtils.java @@ -59,7 +59,7 @@ public static File createSavepointWithOperatorState( new CheckpointMetadata(savepointId, operatorStates, Collections.emptyList()); try (FileOutputStream fileOutputStream = new FileOutputStream(savepointFile)) { - Checkpoints.storeCheckpointMetadata(savepoint, fileOutputStream); + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir(savepoint, fileOutputStream); } return savepointFile; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java index 0bcf91e065435d..ffa8e2127562b6 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java @@ -963,7 +963,7 @@ private static byte[] serializeAsCheckpointMetadata(OperatorID id, byte[] coordi 1337L, Collections.singletonList(state), Collections.emptyList()); final ByteArrayOutputStream out = new ByteArrayOutputStream(); - Checkpoints.storeCheckpointMetadata(metadata, out); + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir(metadata, out); return out.toByteArray(); } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBClaimSavepointThenCheckpointRestoreTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBClaimSavepointThenCheckpointRestoreTest.java new file mode 100644 index 00000000000000..4115ef85beeb42 --- /dev/null +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBClaimSavepointThenCheckpointRestoreTest.java @@ -0,0 +1,354 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.rocksdb; + +import org.apache.flink.api.common.state.ValueState; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.core.execution.SavepointFormatType; +import org.apache.flink.runtime.checkpoint.CheckpointOptions; +import org.apache.flink.runtime.checkpoint.Checkpoints; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.checkpoint.SavepointType; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.state.CheckpointStorageLocationReference; +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.KeyedStateHandle; +import org.apache.flink.runtime.state.PlaceholderStreamStateHandle; +import org.apache.flink.runtime.state.SharedStateRegistry; +import org.apache.flink.runtime.state.SharedStateRegistryImpl; +import org.apache.flink.runtime.state.SnapshotResult; +import org.apache.flink.runtime.state.VoidNamespace; +import org.apache.flink.runtime.state.VoidNamespaceSerializer; +import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess; +import org.apache.flink.runtime.state.filesystem.FileStateHandle; +import org.apache.flink.runtime.state.filesystem.FsCheckpointStreamFactory; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; +import org.apache.flink.testutils.junit.utils.TempDirUtils; +import org.apache.flink.util.IOUtils; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.RunnableFuture; +import java.util.stream.Collectors; + +import static org.apache.flink.core.fs.Path.fromLocalFile; +import static org.apache.flink.core.fs.local.LocalFileSystem.getSharedInstance; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies that checkpoint metadata keeps its file references self-consistent when a checkpoint's + * shared state contains files that live outside the checkpoint's own exclusive directory. + * + *

A NATIVE savepoint writes its SST files into the savepoint directory as {@link + * RelativeFileStateHandle}s (see {@code RocksIncrementalSnapshotStrategy}, which writes savepoint + * SSTs with the EXCLUSIVE state scope but stores them in the {@code sharedState} collection of the + * {@link IncrementalRemoteKeyedStateHandle}). A backend restored from such a savepoint (as happens + * when a job is resumed from it in CLAIM mode) reuses those files, so the first incremental + * checkpoint taken afterwards references relative handles that resolve against the savepoint + * directory rather than the checkpoint's own exclusive directory. + * + *

On deserialization, a relative handle is resolved against the directory of the metadata being + * read ({@code new Path(exclusiveDirPath, relativePath)}). The metadata serializer ({@code + * MetadataV2V3SerializerBase}) must therefore use the relative encoding only for handles that + * actually belong to the exclusive directory being written, and persist the absolute path for all + * others: a foreign handle written as relative would resolve to {@code checkpointDir/} while + * the bytes live at {@code savepointDir/}, and restoring from the checkpoint would fail with + * a file-not-found error. + * + *

This test verifies the behavior end-to-end: it takes a NATIVE savepoint, restores a backend + * from it, takes the first incremental checkpoint, round-trips that checkpoint's metadata through + * an on-disk metadata file, and restores a RocksDB backend from the reloaded handles. + */ +class RocksDBClaimSavepointThenCheckpointRestoreTest { + + private static final int NUM_KEYS = 1000; + private static final int NUM_KEY_GROUPS = 2; + private static final KeyGroupRange KEY_GROUP_RANGE = new KeyGroupRange(0, 1); + + @TempDir private java.nio.file.Path tmp; + + @Test + void testFirstIncrementalCheckpointAfterClaimingNativeSavepointSurvivesMetadataRoundTrip() + throws Exception { + + final ValueStateDescriptor stateDescriptor = + new ValueStateDescriptor<>("state", String.class); + + // ----------------------------------------------------------------------------------- + // (1) backend1: write state, take a normal incremental checkpoint to force real SSTs, + // write more state, then take a NATIVE savepoint. + // ----------------------------------------------------------------------------------- + final File savepointDir = TempDirUtils.newFolder(tmp, "savepoint"); + final File savepointSharedDir = TempDirUtils.newFolder(tmp, "savepoint-shared"); + final File cp1Dir = TempDirUtils.newFolder(tmp, "cp1"); + final File cp1SharedDir = TempDirUtils.newFolder(tmp, "cp1-shared"); + + final KeyedStateHandle savepointHandle; + RocksDBKeyedStateBackend backend1 = + buildBackend(TempDirUtils.newFolder(tmp, "backend1"), Collections.emptyList()); + try { + writeKeys(backend1, stateDescriptor, 0, NUM_KEYS); + // A normal incremental checkpoint forces RocksDB to flush memtables into SST files. + takeSnapshot( + backend1, + 1L, + streamFactory(cp1Dir, cp1SharedDir), + CheckpointOptions.forCheckpointWithDefaultLocation()); + // Write some more so the savepoint contains non-trivial state. + writeKeys(backend1, stateDescriptor, 0, NUM_KEYS); + + savepointHandle = + takeSnapshot( + backend1, + 2L, + streamFactory(savepointDir, savepointSharedDir), + new CheckpointOptions( + SavepointType.savepoint(SavepointFormatType.NATIVE), + CheckpointStorageLocationReference.getDefault())); + } finally { + IOUtils.closeQuietly(backend1); + backend1.dispose(); + } + + // Precondition: the NATIVE savepoint stores its SSTs as relative handles in shared state. + assertThat(savepointHandle).isInstanceOf(IncrementalRemoteKeyedStateHandle.class); + assertThat(((IncrementalRemoteKeyedStateHandle) savepointHandle).getSharedState()) + .as("NATIVE savepoint should write its SST files as RelativeFileStateHandles") + .anyMatch( + handleAndPath -> + handleAndPath.getHandle() instanceof RelativeFileStateHandle); + + // ----------------------------------------------------------------------------------- + // (2) backend2: restore from the savepoint (CLAIM-mode reuse at this level) and take the + // first incremental checkpoint to a DIFFERENT checkpoint directory. + // ----------------------------------------------------------------------------------- + final File checkpointDir = TempDirUtils.newFolder(tmp, "cp2"); + final File checkpointSharedDir = TempDirUtils.newFolder(tmp, "cp2-shared"); + + final IncrementalRemoteKeyedStateHandle checkpointHandle; + RocksDBKeyedStateBackend backend2 = + buildBackend( + TempDirUtils.newFolder(tmp, "backend2"), + Collections.singletonList(savepointHandle)); + try { + checkpointHandle = + (IncrementalRemoteKeyedStateHandle) + takeSnapshot( + backend2, + 3L, + streamFactory(checkpointDir, checkpointSharedDir), + CheckpointOptions.forCheckpointWithDefaultLocation()); + } finally { + IOUtils.closeQuietly(backend2); + backend2.dispose(); + } + + // Precondition: the first checkpoint reuses the savepoint's SST files. They show up as + // placeholders that will resolve, via the SharedStateRegistry, to whatever was registered + // for those files during the CLAIM-mode restore of the savepoint. + assertThat(checkpointHandle.getSharedState()) + .as( + "First incremental checkpoint after claiming the savepoint should reuse the " + + "savepoint's SSTs (carried as placeholders in shared state)") + .anyMatch( + handleAndPath -> + handleAndPath.getHandle() instanceof PlaceholderStreamStateHandle); + + // ----------------------------------------------------------------------------------- + // (3) Register shared state as the JobManager does (the relevant subset): + // a) register the claimed savepoint on CLAIM-mode restore + // (SharedStateRegistry#registerAllAfterRestored -> registerSharedStates), then + // b) register the first completed incremental checkpoint, whose placeholders now + // resolve against the entries registered in (a). + // After this step the checkpoint's shared state holds relative handles that resolve + // against the savepoint directory, which is what the metadata serializer sees when the + // JobManager finalizes the checkpoint. + // ----------------------------------------------------------------------------------- + final SharedStateRegistry registry = new SharedStateRegistryImpl(); + IncrementalRemoteKeyedStateHandle savepointIncremental = + (IncrementalRemoteKeyedStateHandle) savepointHandle; + savepointIncremental.registerSharedStates(registry, savepointIncremental.getCheckpointId()); + checkpointHandle.registerSharedStates(registry, 3L); + + final IncrementalRemoteKeyedStateHandle restoredHandle = + metadataRoundTrip(checkpointHandle, checkpointDir, 3L); + + // RocksDB-independent assertion: EVERY shared file referenced by the reloaded + // checkpoint must physically exist on disk. This covers the checkpoint's own new SSTs + // (absolute handles in the shared-state directory) as well as the inherited savepoint + // SSTs (foreign handles that must keep their absolute location). If a foreign handle were + // written with the relative encoding, it would be resolved against checkpointDir/ + // (which does not exist) instead of savepointDir/. + List referencedSharedFiles = + restoredHandle.getSharedState().stream() + .map(HandleAndLocalPath::getHandle) + .filter(FileStateHandle.class::isInstance) + .map(handle -> new File(((FileStateHandle) handle).getFilePath().getPath())) + .collect(Collectors.toList()); + assertThat(referencedSharedFiles) + .as("The reloaded checkpoint must reference at least one shared file on disk") + .isNotEmpty() + .allSatisfy(file -> assertThat(file).exists()); + + // ----------------------------------------------------------------------------------- + // (4) backend3: restore from the reloaded checkpoint handle and verify the state. + // ----------------------------------------------------------------------------------- + RocksDBKeyedStateBackend backend3 = + buildBackend( + TempDirUtils.newFolder(tmp, "backend3"), + Collections.singletonList(restoredHandle)); + try { + assertStateRestored(backend3, stateDescriptor, 0, NUM_KEYS); + } finally { + IOUtils.closeQuietly(backend3); + backend3.dispose(); + } + } + + // ------------------------------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------------------------------ + + private RocksDBKeyedStateBackend buildBackend( + File instanceBasePath, Collection restoreHandles) throws Exception { + return RocksDBTestUtils.builderForTestDefaults( + instanceBasePath, + IntSerializer.INSTANCE, + NUM_KEY_GROUPS, + KEY_GROUP_RANGE, + restoreHandles) + .setEnableIncrementalCheckpointing(true) + .build(); + } + + private FsCheckpointStreamFactory streamFactory(File checkpointDir, File sharedStateDir) { + return new FsCheckpointStreamFactory( + getSharedInstance(), + fromLocalFile(checkpointDir), + fromLocalFile(sharedStateDir), + 1, + 4096); + } + + private void writeKeys( + RocksDBKeyedStateBackend backend, + ValueStateDescriptor descriptor, + int fromInclusive, + int toExclusive) + throws Exception { + for (int key = fromInclusive; key < toExclusive; key++) { + backend.setCurrentKey(key); + ValueState state = + backend.getPartitionedState( + VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, descriptor); + state.update(valueFor(key)); + } + } + + private void assertStateRestored( + RocksDBKeyedStateBackend backend, + ValueStateDescriptor descriptor, + int fromInclusive, + int toExclusive) + throws Exception { + for (int key = fromInclusive; key < toExclusive; key++) { + backend.setCurrentKey(key); + ValueState state = + backend.getPartitionedState( + VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, descriptor); + assertThat(state.value()).isEqualTo(valueFor(key)); + } + } + + private static String valueFor(int key) { + // Reasonably long values so that RocksDB produces real SST files. + return "value-for-key-" + key + "-0123456789abcdefghijklmnopqrstuvwxyz"; + } + + private static KeyedStateHandle takeSnapshot( + RocksDBKeyedStateBackend backend, + long checkpointId, + FsCheckpointStreamFactory streamFactory, + CheckpointOptions options) + throws Exception { + RunnableFuture> snapshot = + backend.snapshot(checkpointId, checkpointId, streamFactory, options); + snapshot.run(); + return snapshot.get().getJobManagerOwnedSnapshot(); + } + + /** + * Serializes the given handle into checkpoint metadata stored in {@code exclusiveDir} under + * {@link AbstractFsCheckpointStorageAccess#METADATA_FILE_NAME} and reloads it, mirroring what + * the JobManager does. Serialization is told the checkpoint's exclusive directory (as + * production does via {@code FsCheckpointMetadataOutputStream}), and deserialization uses + * {@code exclusiveDir} as the external pointer, so any {@link RelativeFileStateHandle} that + * actually belongs to {@code exclusiveDir} is resolved relative to it while foreign relative + * handles are written (and read back) as absolute. + */ + private static IncrementalRemoteKeyedStateHandle metadataRoundTrip( + IncrementalRemoteKeyedStateHandle handle, File exclusiveDir, long checkpointId) + throws Exception { + OperatorSubtaskState subtaskState = + OperatorSubtaskState.builder().setManagedKeyedState(handle).build(); + OperatorState operatorState = new OperatorState(null, null, new OperatorID(), 1, 2); + operatorState.putState(0, subtaskState); + CheckpointMetadata metadata = + new CheckpointMetadata( + checkpointId, + Collections.singletonList(operatorState), + Collections.emptyList()); + + File metadataFile = + new File(exclusiveDir, AbstractFsCheckpointStorageAccess.METADATA_FILE_NAME); + try (DataOutputStream dos = + new DataOutputStream(new FileOutputStream(metadataFile, false))) { + Checkpoints.storeCheckpointMetadata(metadata, dos, fromLocalFile(exclusiveDir)); + } + + CheckpointMetadata reloaded; + try (DataInputStream dis = new DataInputStream(new FileInputStream(metadataFile))) { + reloaded = + Checkpoints.loadCheckpointMetadata( + dis, + RocksDBClaimSavepointThenCheckpointRestoreTest.class.getClassLoader(), + exclusiveDir.getAbsolutePath()); + } + + OperatorState reloadedOperatorState = reloaded.getOperatorStates().iterator().next(); + OperatorSubtaskState reloadedSubtaskState = + reloadedOperatorState.getStates().iterator().next(); + return (IncrementalRemoteKeyedStateHandle) + reloadedSubtaskState.getManagedKeyedState().iterator().next(); + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointAfterClaimedNativeSavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointAfterClaimedNativeSavepointITCase.java new file mode 100644 index 00000000000000..eefad72b9720c1 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointAfterClaimedNativeSavepointITCase.java @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.checkpointing; + +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.functions.OpenContext; +import org.apache.flink.api.common.functions.RichMapFunction; +import org.apache.flink.api.common.state.ValueState; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.client.program.ClusterClient; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ExternalizedCheckpointRetention; +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.configuration.RestartStrategyOptions; +import org.apache.flink.configuration.StateBackendOptions; +import org.apache.flink.configuration.StateChangelogOptions; +import org.apache.flink.core.execution.RecoveryClaimMode; +import org.apache.flink.core.execution.SavepointFormatType; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; +import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; +import org.apache.flink.runtime.state.filesystem.FileStateHandle; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.state.rocksdb.RocksDBConfigurableOptions; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; +import org.apache.flink.test.util.MiniClusterWithClientResource; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning; +import static org.apache.flink.test.util.TestUtils.loadCheckpointMetadata; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies, through a real cluster, that an incremental checkpoint taken after a CLAIM-mode restore + * from a NATIVE savepoint remains restorable. + * + *

A NATIVE RocksDB savepoint references its SST files relative to the savepoint directory. A job + * restored from it in CLAIM mode reuses those files, so its first incremental checkpoint references + * files that live outside the checkpoint's own exclusive directory. The checkpoint metadata must + * record such references by their absolute location: a relative encoding would be resolved against + * the checkpoint directory when the metadata is read back, and restoring from the checkpoint would + * fail looking for files that were never there. + * + *

{@code RocksDBClaimSavepointThenCheckpointRestoreTest} pins the same property at the + * backend/serializer level, in-process, without the JobManager finalization wiring. This test + * instead drives the whole path on a real cluster, so it also covers what the component test + * cannot: in particular that the metadata finalization in {@code PendingCheckpoint} binds to the + * {@code CheckpointMetadataOutputStream} overload of {@code Checkpoints#storeCheckpointMetadata}, + * which hands the checkpoint's exclusive directory to the serializer. + */ +class ResumeCheckpointAfterClaimedNativeSavepointITCase { + + private static final int PARALLELISM = 2; + private static final int NUM_KEYS = 128; + + @TempDir private Path checkpointsDir; + @TempDir private Path savepointsDir; + + @Test + void testRestoreFromFirstIncrementalCheckpointAfterClaimingNativeSavepoint() throws Exception { + final Configuration config = new Configuration(); + config.set(StateBackendOptions.STATE_BACKEND, "rocksdb"); + config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true); + // Force the changelog backend and checkpoint file merging off: the scenario requires the + // savepoint SSTs to be plain (relative) file references reused by the next incremental + // checkpoint, and the test-infra config randomization would otherwise wrap the state in + // changelog handles or replace the file references with file-merging segments. Aside from + // the RocksDB ingest-restore mode pinned below, the remaining randomized options do not + // affect the keyed shared state this test asserts on (unaligned checkpoints, if + // randomized on, at most add channel-state files). + config.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, false); + config.set(CheckpointingOptions.FILE_MERGING_ENABLED, false); + // Pinned for determinism: the test infra randomizes this per commit. It is inert for the + // same-parallelism single-handle restores in this test, but a future rescaling variant + // would silently start depending on the randomized value. + config.set(RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE, false); + config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointsDir.toUri().toString()); + config.set( + CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION, + ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION); + // Write every state file as a real file (nothing inlined into the metadata) so the + // shared-file assertion below sees actual file paths. + config.set(CheckpointingOptions.FS_SMALL_FILE_THRESHOLD, MemorySize.ZERO); + // A restore failure must surface as a terminal job failure instead of hiding in an + // endless restart loop, so that this test fails fast with the actual cause. + config.set(RestartStrategyOptions.RESTART_STRATEGY, "none"); + + // Manual cluster lifecycle (instead of MiniClusterExtension) because the configuration + // depends on per-test @TempDirs; same pattern as SavepointFormatITCase. + final MiniClusterWithClientResource cluster = + new MiniClusterWithClientResource( + new MiniClusterResourceConfiguration.Builder() + .setConfiguration(config) + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(PARALLELISM) + .build()); + cluster.before(); + try { + final ClusterClient client = cluster.getClusterClient(); + final MiniCluster miniCluster = cluster.getMiniCluster(); + + // (1) Trigger a checkpoint before the savepoint so the savepoint's SSTs are a + // realistic mix of reused and freshly flushed files, then stop with a NATIVE + // savepoint (SST references relative to the savepoint directory). + final JobGraph initialJob = createJobGraph(config); + client.submitJob(initialJob).get(); + waitForAllTaskRunning(miniCluster, initialJob.getJobID(), false); + miniCluster.triggerCheckpoint(initialJob.getJobID()).get(); + final String savepointPath = + client.stopWithSavepoint( + initialJob.getJobID(), + false, + savepointsDir.toUri().toString(), + SavepointFormatType.NATIVE) + .get(); + + // (2) Restore in CLAIM mode and take the first incremental checkpoint. It reuses the + // savepoint's SSTs, so its metadata references files outside its own directory. + final JobGraph restoredFromSavepoint = createJobGraph(config); + restoredFromSavepoint.setSavepointRestoreSettings( + SavepointRestoreSettings.forPath( + savepointPath, false, RecoveryClaimMode.CLAIM)); + client.submitJob(restoredFromSavepoint).get(); + waitForAllTaskRunning(miniCluster, restoredFromSavepoint.getJobID(), false); + final String checkpointPath = + miniCluster.triggerCheckpoint(restoredFromSavepoint.getJobID()).get(); + client.cancel(restoredFromSavepoint.getJobID()).get(); + miniCluster.requestJobResult(restoredFromSavepoint.getJobID()).get(); + + // The startsWith check is the regression guard for absolute-vs-relative encoding + // (see class javadoc); isNotEmpty() only separates a broken setup (no file-backed + // shared state at all) from a real regression, since anySatisfy itself already fails + // on an empty list. The trailing separator keeps a sibling directory sharing the + // prefix from matching. + final String savepointPrefix = + savepointPath.endsWith("/") ? savepointPath : savepointPath + "/"; + assertThat(sharedStateFilePaths(checkpointPath)) + .as( + "the first incremental checkpoint after claiming the savepoint must " + + "reference reused savepoint files by their absolute location") + .isNotEmpty() + .anySatisfy(path -> assertThat(path).startsWith(savepointPrefix)); + + // (3) Restore from that checkpoint. Restarts are disabled, so a missing-file failure + // goes terminal and waitForAllTaskRunning surfaces the cause instead of retrying + // silently. The restored job must then complete one more checkpoint on its own. + final JobGraph restoredFromCheckpoint = createJobGraph(config); + restoredFromCheckpoint.setSavepointRestoreSettings( + SavepointRestoreSettings.forPath(checkpointPath, false)); + client.submitJob(restoredFromCheckpoint).get(); + waitForAllTaskRunning(miniCluster, restoredFromCheckpoint.getJobID(), false); + miniCluster.triggerCheckpoint(restoredFromCheckpoint.getJobID()).get(); + client.cancel(restoredFromCheckpoint.getJobID()).get(); + miniCluster.requestJobResult(restoredFromCheckpoint.getJobID()).get(); + } finally { + cluster.after(); + } + } + + /** All absolute file paths referenced by the checkpoint's keyed shared state. */ + private static List sharedStateFilePaths(String checkpointPath) throws Exception { + final CheckpointMetadata metadata = loadCheckpointMetadata(checkpointPath); + return metadata.getOperatorStates().stream() + .flatMap(operatorState -> operatorState.getStates().stream()) + .flatMap(subtaskState -> subtaskState.getManagedKeyedState().stream()) + .filter(IncrementalRemoteKeyedStateHandle.class::isInstance) + .map(IncrementalRemoteKeyedStateHandle.class::cast) + .flatMap(handle -> handle.getSharedState().stream()) + .map(HandleAndLocalPath::getHandle) + .filter(FileStateHandle.class::isInstance) + .map(handle -> ((FileStateHandle) handle).getFilePath().toString()) + .collect(Collectors.toList()); + } + + private static JobGraph createJobGraph(Configuration config) { + final StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(config); + env.setParallelism(PARALLELISM); + + // The throttle is chained to the source, upstream of the keyBy exchange, so it limits the + // production rate itself: checkpoint barriers never queue behind a large record backlog, + // which keeps every aligned checkpoint and the savepoint fast. + // Effectively infinite, but deliberately not the full Long range: splitting + // Long.MIN_VALUE..Long.MAX_VALUE overflows in NumberSequenceIterator#split and yields + // overlapping full-range splits instead of a partition. + env.fromSequence(0, Long.MAX_VALUE - 1) + .map(new Throttler()) + .keyBy(value -> Math.floorMod(value, NUM_KEYS)) + .map(new StatefulCounter()) + .uid("stateful-counter") + .sinkTo(new DiscardingSink<>()); + + return env.getStreamGraph().getJobGraph(); + } + + /** + * Slows record production so that little new state is written between the restore and the + * post-restore checkpoint, keeping RocksDB from compacting away the SST files reused from the + * claimed savepoint. Best-effort: if reuse stops happening, the shared-file assertion fails. + */ + private static final class Throttler implements MapFunction { + + private static final long THROTTLE_MILLIS = 1; + + @Override + public Long map(Long value) throws Exception { + Thread.sleep(THROTTLE_MILLIS); + return value; + } + } + + /** Keyed counter whose RocksDB state makes the savepoint and checkpoints non-trivial. */ + private static final class StatefulCounter extends RichMapFunction { + + private ValueState counter; + + @Override + public void open(OpenContext openContext) throws Exception { + counter = + getRuntimeContext() + .getState( + new ValueStateDescriptor<>( + "counter", BasicTypeInfo.LONG_TYPE_INFO)); + } + + @Override + public Long map(Long value) throws Exception { + long next = Optional.ofNullable(counter.value()).orElse(0L) + value; + counter.update(next); + return next; + } + } +}