From fbe2e6bb398f3d000d454e5bb2051cb1507828c7 Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Fri, 3 Jul 2026 19:24:56 +0200 Subject: [PATCH 1/5] [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 e84e8bb1bc1cc70e1bfc2788f842b8c4cf311f95 Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Wed, 8 Jul 2026 17:59:11 +0200 Subject: [PATCH 2/5] [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. Co-Authored-By: Claude Fable 5 --- .../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 | 151 ++++-- .../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 | 183 ++++++++ .../MetadataV2V3SerializerBaseTest.java | 437 ++++++++++++++++++ .../flink/runtime/jobmaster/TestUtils.java | 2 +- .../OperatorCoordinatorSchedulerTest.java | 2 +- ...aimSavepointThenCheckpointRestoreTest.java | 350 ++++++++++++++ ...ointAfterClaimedNativeSavepointITCase.java | 238 ++++++++++ 21 files changed, 1716 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..003c45e575b8e4 --- /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}. {@code + * SavepointDeepCopyTest} covers the same guarantee end-to-end on a cluster, where a regression + * surfaces only as a restore failure far from its cause; this test pins the encoding decision + * itself, so a regression fails fast and points at the responsible code. + */ +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 { + File sourceSavepointDir = new File(tempDir, "sp-old"); + File targetSavepointDir = new File(tempDir, "sp-new"); + assertThat(sourceSavepointDir.mkdirs()).isTrue(); + assertThat(targetSavepointDir.mkdirs()).isTrue(); + + // 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); + + // Copy the retained file the way SavepointWriter.fromExistingSavepoint does: through + // FileCopyFunction, which copies by bare file name into the new savepoint directory. The + // metadata written below is self-contained only together with this naming contract. + FileCopyFunction copyFunction = new FileCopyFunction(targetSavepointDir.getAbsolutePath()); + copyFunction.open(FirstAttemptInitializationContext.of(SUBTASK_INDEX, PARALLELISM)); + copyFunction.writeRecord(sourceStatePath); + copyFunction.close(); + + CheckpointMetadata metadata = createMetadata(retainedHandle); + + SavepointOutputFormat format = new SavepointOutputFormat(targetSavepointPath); + format.setRuntimeContext(new MockStreamingRuntimeContext(PARALLELISM, SUBTASK_INDEX)); + format.open(FirstAttemptInitializationContext.of(SUBTASK_INDEX, PARALLELISM)); + format.writeRecord(metadata); + format.close(); + + // Reload the written metadata; relative handles resolve against the directory the + // metadata is read from, i.e. the new savepoint holding the copies. + CheckpointMetadata reloaded = + SavepointLoader.loadSavepointMetadata(targetSavepointPath.getPath()); + StreamStateHandle reloadedShared = extractSingleSharedStateHandle(reloaded); + + 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..23205087540029 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,26 @@ 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)) { + // The relative encoding stores just the file name; on restore it is resolved + // against the directory the metadata is read from. So it is only used for files + // that live in the directory being written. Handles to files anywhere else fail + // canKeepRelativeEncoding and take the FileStateHandle branch below, which keeps + // their full 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 +803,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 +864,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 +955,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 +986,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 +1024,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..7989b187e21bcc --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java @@ -0,0 +1,183 @@ +/* + * 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 that the {@link MetadataSerializer#serialize} entry points agree with each other across + * every registered metadata version: + * + *

    + *
  • without an exclusive directory, every entry point produces byte-identical output — a + * convenience overload must not drift from the contract-bearing method it delegates to; + *
  • a supplied exclusive directory reaches the relative-handle encoder instead of being + * silently dropped: making a shared relative handle foreign to it changes the written bytes; + *
  • the deserialize-only versions reject serialization through every entry point. + *
+ * + *

Which method carries the contract and how the convenience overload delegates to it is + * documented on {@link MetadataSerializer#serialize(CheckpointMetadata, DataOutputStream, Path)}. + */ +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..90965496f5c26a --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBaseTest.java @@ -0,0 +1,437 @@ +/* + * 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. + * + *

Throughout this test an own handle is one whose file lives in the directory the + * metadata is written to; a foreign handle points anywhere else. + */ +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"); + writeEmptyMetadataFile(checkpointDir); + + final Path exclusiveDirPath = Path.fromLocalFile(checkpointDir); + final Path savepointDirPath = Path.fromLocalFile(savepointDir); + + final String relativeFileName = UUID.randomUUID().toString(); + 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"); + 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 (a {@code null} {@code exclusiveDirPath}, which is what + * the state processor API requests via {@code + * Checkpoints#storeCheckpointMetadataWithoutExclusiveDir}), 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(); + } + + /** + * Creates an empty {@code _metadata} file so the directory resolves via {@code + * AbstractFsCheckpointStorageAccess#resolveCheckpointPointer}, which deserialization uses to + * rebuild relative handles. This keeps a wrongly-relative handle failing on the test's own + * assertion instead of on an IOException about a missing _metadata file. + */ + 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..02a2f7beccbb9a --- /dev/null +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBClaimSavepointThenCheckpointRestoreTest.java @@ -0,0 +1,350 @@ +/* + * 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.nio.file.Path; +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 whose files live in the savepoint + * directory, not in 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 full round trip in process: 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. The + * JobManager wiring it bypasses (metadata finalization through {@code + * CheckpointMetadataOutputStream}) is covered by {@code + * ResumeCheckpointAfterClaimedNativeSavepointITCase} in flink-tests. + */ +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 Path tmp; + + @Test + void testFirstCheckpointSurvivesMetadataRoundTrip() throws Exception { + + final ValueStateDescriptor stateDescriptor = + new ValueStateDescriptor<>("state", String.class); + + // 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()); + // More writes on top of the flushed SSTs, so the savepoint covers both previously + // flushed files and fresh memtable data. + 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); + + // 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); + + // 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 pointing at + // files in 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 any SSTs the checkpoint + // uploaded itself (absolute handles in the shared-state directory) as well as the + // inherited savepoint SSTs. In this scenario nothing was written after the restore, + // so the shared state holds only the reused savepoint files. If such a 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()); + + // 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) { + 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..ae4ae9f5530908 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointAfterClaimedNativeSavepointITCase.java @@ -0,0 +1,238 @@ +/* + * 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 testFirstCheckpointAfterClaimIsRestorable() throws Exception { + final Configuration config = new Configuration(); + config.set(StateBackendOptions.STATE_BACKEND, "rocksdb"); + config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true); + config.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, false); + config.set(CheckpointingOptions.FILE_MERGING_ENABLED, false); + 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); + // Keep every state file a real file so the shared-file assertion sees actual paths. + config.set(CheckpointingOptions.FS_SMALL_FILE_THRESHOLD, MemorySize.ZERO); + // Fail fast on a broken restore instead of looping through restarts. + config.set(RestartStrategyOptions.RESTART_STRATEGY, "none"); + + // Manual lifecycle because the configuration depends on per-test @TempDirs. + 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(); + + // Checkpoint once so the savepoint covers a mix of reused and freshly flushed SSTs, + // then stop with a NATIVE savepoint. + 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(); + + // Restore in CLAIM mode, the first incremental checkpoint reuses the savepoint's SSTs. + 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 regression guard: reused savepoint files must be recorded by their absolute + // location. The trailing separator keeps a sibling directory 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)); + + // Restore from that checkpoint (a missing-file failure would go terminal and surface + // here) and let the restored job complete one more checkpoint of 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, so barriers never queue behind a record backlog. + // Not the full Long range: splitting that overflows in NumberSequenceIterator#split. + 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 production so the reused savepoint SSTs are not compacted away before the post-restore + * checkpoint. 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; + } + } + + /** Gives the job keyed RocksDB state, without it there would be no shared SSTs to assert on. */ + 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; + } + } +} From 70dff60af58276a15ed5b87687ce24c989e4600d Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Wed, 15 Jul 2026 12:27:34 +0200 Subject: [PATCH 3/5] [FLINK-39964][state] Explain the exclusive directory and elaborate about the metadata store variants --- .../flink/runtime/checkpoint/Checkpoints.java | 35 +++++++++++++------ .../metadata/MetadataSerializer.java | 32 ++++++++--------- 2 files changed, 41 insertions(+), 26 deletions(-) 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 91c856c184dde7..b167ed401f1f80 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 @@ -32,10 +32,12 @@ 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.CheckpointedStateScope; import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation; import org.apache.flink.runtime.state.StateBackend; import org.apache.flink.runtime.state.StateBackendLoader; import org.apache.flink.runtime.state.StreamStateHandle; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage; import org.apache.flink.util.CollectionUtil; @@ -66,6 +68,14 @@ * 

[MagicNumber (int) | Format Version (int) | Checkpoint Metadata (variable)]
* *

The actual savepoint serialization is version-specific via the {@link MetadataSerializer}. + * + *

Checkpoint metadata may reference state files through {@link RelativeFileStateHandle}s, which + * store only a file name. On recovery such a reference is resolved against the checkpoint's + * exclusive directory: the per-checkpoint directory that contains the {@code _metadata} file + * and the checkpoint's own, non-shared state files, e.g. {@code /checkpoints//chk-42/} (see + * {@link CheckpointedStateScope#EXCLUSIVE}). This yields the correct path only for files that + * actually live in that directory, which is why the {@code storeCheckpointMetadata} variants below + * differ in how they encode relative handles. */ public class Checkpoints { @@ -79,15 +89,16 @@ public class Checkpoints { // ------------------------------------------------------------------------ /** - * 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, + * Stores the checkpoint metadata with every {@link RelativeFileStateHandle} kept relative + * unconditionally, regardless of where its file currently lives. This is only correct if the + * caller guarantees that every referenced file is, or will be, located in the directory the + * metadata is eventually read from. The state processor API establishes that guarantee by + * copying the referenced files next to the new metadata (see {@code SavepointOutputFormat}); + * everything else should use {@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. + *

The deliberately different method name (rather than an overload) makes choosing this + * encoding an explicit decision at the call site. */ public static void storeCheckpointMetadataWithoutExclusiveDir( CheckpointMetadata checkpointMetadata, OutputStream out) throws IOException { @@ -97,9 +108,13 @@ public static void storeCheckpointMetadataWithoutExclusiveDir( } /** - * 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. + * Stores the checkpoint metadata, letting the serializer check every {@link + * RelativeFileStateHandle} against the exclusive directory the stream writes into ({@link + * CheckpointMetadataOutputStream#getExclusiveCheckpointDir()}): a handle whose file lives in + * that directory keeps the relative encoding, while a handle pointing anywhere else (e.g. an + * SST file reused from a claimed savepoint) is persisted with its absolute path, because the + * relative form would be re-anchored to the wrong directory when the metadata is read back. + * Use this variant when writing an actual checkpoint or savepoint. */ public static void storeCheckpointMetadata( CheckpointMetadata checkpointMetadata, CheckpointMetadataOutputStream out) 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 04c551e8db0e4c..c80108ca2d3624 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 @@ -20,6 +20,7 @@ import org.apache.flink.core.fs.Path; import org.apache.flink.core.io.Versioned; +import org.apache.flink.runtime.checkpoint.Checkpoints; import javax.annotation.Nullable; @@ -49,13 +50,11 @@ CheckpointMetadata deserialize( throws IOException; /** - * 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. + * Convenience variant of {@link #serialize(CheckpointMetadata, DataOutputStream, Path)} with a + * {@code null} exclusive directory: every relative file reference is kept relative, + * unconditionally, and is resolved against whatever directory the metadata is later read from. + * Only correct if every referenced file ends up in that directory; see {@link + * Checkpoints#storeCheckpointMetadataWithoutExclusiveDir} for the use case. * * @throws IOException Serialization failures are forwarded */ @@ -65,21 +64,22 @@ default void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream d } /** - * 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. + * Serializes a savepoint or checkpoint metadata to an output stream. * - *

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. + *

Implementations check every relative file reference against {@code exclusiveDirPath}: + * references to files inside that directory keep the relative encoding, references to files + * anywhere else (e.g. a savepoint SST reused by the first incremental checkpoint after a + * CLAIM-mode restore) are persisted with their absolute path, because the relative form would + * be resolved against the wrong directory on recovery. See {@link Checkpoints} for what the + * exclusive directory is. * *

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. + * @param exclusiveDirPath the directory that will contain the metadata file (the checkpoint's + * exclusive directory), or {@code null} to keep every relative reference relative + * regardless of where its file lives * @throws IOException Serialization failures are forwarded */ void serialize( From 6be50ec12534db4fe138f2b981907b6c39410248 Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Wed, 15 Jul 2026 12:27:34 +0200 Subject: [PATCH 4/5] [FLINK-39964][state-processor-api] Explain the deep-copy pipeline behind SavepointOutputFormat's encoding choice --- .../api/output/SavepointOutputFormat.java | 23 ++++++++++--------- 1 file changed, 12 insertions(+), 11 deletions(-) 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 2689479763d5e8..fd10a0e455700e 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,17 +79,18 @@ public void writeRecord(CheckpointMetadata metadata) throws IOException { () -> { try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) { - // 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. + // Must stay on the variant WITHOUT the exclusive directory. + // SavepointWriter#fromExistingSavepoint assembles a new savepoint + // from an existing one: state handles of unmodified operators are + // carried over as-is, still pointing into the source savepoint, + // while a FileCopyFunction in the same job copies their files + // into this savepoint's directory under the same file name. + // Writing every relative reference as just the file name makes + // it resolve to the local copy on restore, so this savepoint stays + // self-contained even after the source savepoint is deleted. The + // exclusive-dir-aware Checkpoints.storeCheckpointMetadata would + // instead record the carried-over handles' absolute paths, which + // still point into the source savepoint. Checkpoints.storeCheckpointMetadataWithoutExclusiveDir( metadata, out); CompletedCheckpointStorageLocation finalizedLocation = From 30c2e72cd3a2b00ec1b06e2480ac380c660bfca2 Mon Sep 17 00:00:00 2001 From: Aleksandr Savonin Date: Wed, 15 Jul 2026 12:27:35 +0200 Subject: [PATCH 5/5] [FLINK-39964][state] Make every serializer caller pass its serialization context explicitly --- .../metadata/ChannelStateHandleSerializerV1.java | 4 +++- .../metadata/ChannelStateHandleSerializerV2.java | 4 +++- .../metadata/MetadataV2V3SerializerBase.java | 10 ---------- .../checkpoint/metadata/MetadataV3Serializer.java | 6 +++++- .../checkpoint/metadata/MetadataV3SerializerTest.java | 6 ++++-- 5 files changed, 15 insertions(+), 15 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java index 22eefc63cf06d5..75f52e933d3332 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java @@ -99,7 +99,9 @@ static void serializeChannelStateHandle( dos.writeLong(offset); } dos.writeLong(handle.getStateSize()); - serializeStreamStateHandle(handle.getDelegate(), dos); + // null context: channel state is always written into the current checkpoint's own + // exclusive directory, see MetadataV3Serializer#serializeSubtaskState. + serializeStreamStateHandle(handle.getDelegate(), dos, null); } static > diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java index ba8f8ea6c6ce23..0bd9f693e47ffe 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java @@ -146,7 +146,9 @@ private void serializeMergedChannelStateHandle( dos.writeInt(handle.getSubtaskIndex()); dos.writeLong(handle.getStateSize()); - serializeStreamStateHandle(handle.getDelegate(), dos); + // null context: channel state is always written into the current checkpoint's own + // exclusive directory, see MetadataV3Serializer#serializeSubtaskState. + serializeStreamStateHandle(handle.getDelegate(), dos, null); dos.writeInt(handle.getSerializedChannelOffsets().length); dos.write(handle.getSerializedChannelOffsets()); 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 23205087540029..6825418f4b84b5 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 @@ -338,11 +338,6 @@ protected OperatorSubtaskState deserializeSubtaskState( // ------------------------------------------------------------------------ @VisibleForTesting - static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputStream dos) - throws IOException { - serializeKeyedStateHandle(stateHandle, dos, null); - } - static void serializeKeyedStateHandle( KeyedStateHandle stateHandle, DataOutputStream dos, @@ -745,11 +740,6 @@ protected void serializeInputStateHandle(InputStateHandle handle, DataOutputStre // low-level state handles // ------------------------------------------------------------------------ - static void serializeStreamStateHandle(StreamStateHandle stateHandle, DataOutputStream dos) - throws IOException { - serializeStreamStateHandle(stateHandle, dos, null); - } - static void serializeStreamStateHandle( StreamStateHandle stateHandle, DataOutputStream dos, 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 c4d7a5be75925c..261b668ea9b578 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 @@ -263,6 +263,10 @@ private void serializeCollection( // ------------------------------------------------------------------------ // exposed static methods for test cases // + // These helpers (de)serialize standalone handles outside any checkpoint, + // so there is no exclusive directory to check relative references + // against: the (de)serialization context is always null. + // // NOTE: The fact that certain tests directly call these lower level // serialization methods is a problem, because that way the tests // bypass the versioning scheme. Especially tests that test for @@ -273,7 +277,7 @@ private void serializeCollection( @VisibleForTesting public static void serializeStreamStateHandle( StreamStateHandle stateHandle, DataOutputStream dos) throws IOException { - MetadataV2V3SerializerBase.serializeStreamStateHandle(stateHandle, dos); + MetadataV2V3SerializerBase.serializeStreamStateHandle(stateHandle, dos, null); } @VisibleForTesting diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3SerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3SerializerTest.java index e3db858ed9eb84..799a28e7826cc3 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3SerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3SerializerTest.java @@ -268,7 +268,8 @@ void testSerializeKeyGroupsStateHandle() throws IOException { try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) { MetadataV2V3SerializerBase.serializeStreamStateHandle( new KeyGroupsStateHandle(offsets, new ByteStreamStateHandle("test", data)), - new DataOutputStream(out)); + new DataOutputStream(out), + null); try (ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray())) { StreamStateHandle handle = MetadataV2V3SerializerBase.deserializeStreamStateHandle( @@ -298,7 +299,8 @@ void testSerializeFullChangelogStateBackendHandle() throws IOException { private void testSerializeChangelogStateBackendHandle(boolean fullSnapshot) throws IOException { ChangelogStateBackendHandle handle = createChangelogStateBackendHandle(fullSnapshot); try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) { - MetadataV2V3SerializerBase.serializeKeyedStateHandle(handle, new DataOutputStream(out)); + MetadataV2V3SerializerBase.serializeKeyedStateHandle( + handle, new DataOutputStream(out), null); try (ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray())) { KeyedStateHandle deserialized = MetadataV2V3SerializerBase.deserializeKeyedStateHandle(