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 144aeb9d1c6dab..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; @@ -173,7 +174,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", @@ -184,6 +185,11 @@ public void testSavepointDeepCopy() throws Exception { stateFiles1, everyItem(isIn(stateFiles2))); + // Not a cleanup step: deleting savepoint1 before reading savepoint2 proves the deep + // copy is self-contained — savepoint2's metadata must reference the copied files, not + // the originals in savepoint1. + FileUtils.deleteDirectory(new File(savepointPath1)); + // Try to fromExistingSavepoint savepoint2 and read the state of "Operator1" (which has not // been // touched/changed when savepoint2 diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java new file mode 100644 index 00000000000000..08b08758d86b06 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.output; + +import org.apache.flink.api.common.io.FirstAttemptInitializationContext; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; +import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.runtime.state.StreamStateHandle; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; +import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; +import org.apache.flink.state.api.runtime.OperatorIDGenerator; +import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.streaming.util.MockStreamingRuntimeContext; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Collections; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that a savepoint written by {@link SavepointOutputFormat} stays self-contained when it + * retains operator state from an existing savepoint, i.e. that it holds no references back into the + * source savepoint it was derived from. + * + *

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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