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 c0d022e04a78b..fd10a0e455700 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,20 @@ 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. + // SavepointWriter#fromExistingSavepoint assembles a new savepoint + // from an existing one: state handles of unmodified operators are + // carried over as-is, still pointing into the source savepoint, + // while a FileCopyFunction in the same job copies their files + // into this savepoint's directory under the same file name. + // Writing every relative reference as just the file name makes + // it resolve to the local copy on restore, so this savepoint stays + // self-contained even after the source savepoint is deleted. The + // exclusive-dir-aware Checkpoints.storeCheckpointMetadata would + // instead record the carried-over handles' absolute paths, which + // still point into the source savepoint. + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir( + metadata, out); CompletedCheckpointStorageLocation finalizedLocation = 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 144aeb9d1c6da..56f3c6f044f9b 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 0000000000000..003c45e575b8e --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/output/SavepointOutputFormatSelfContainedTest.java @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.output; + +import org.apache.flink.api.common.io.FirstAttemptInitializationContext; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; +import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.runtime.state.StreamStateHandle; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; +import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; +import org.apache.flink.state.api.runtime.OperatorIDGenerator; +import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.streaming.util.MockStreamingRuntimeContext; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Collections; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that a savepoint written by {@link SavepointOutputFormat} stays self-contained when it + * retains operator state from an existing savepoint, i.e. that it holds no references back into the + * source savepoint it was derived from. + * + *

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

See the comment in {@link SavepointOutputFormat#writeRecord} for why the metadata must be + * written with {@code Checkpoints.storeCheckpointMetadataWithoutExclusiveDir}. {@code + * SavepointDeepCopyTest} covers the same guarantee end-to-end on a cluster, where a regression + * surfaces only as a restore failure far from its cause; this test pins the encoding decision + * itself, so a regression fails fast and points at the responsible code. + */ +class SavepointOutputFormatSelfContainedTest { + + private static final int PARALLELISM = 1; + private static final int MAX_PARALLELISM = 128; + private static final int SUBTASK_INDEX = 0; + private static final long CHECKPOINT_ID = 1L; + private static final String OPERATOR_UID = "uid"; + private static final String STATE_FILE_NAME = "sst-0001.state"; + + @Test + void testSavepointWithRetainedStateStaysSelfContained(@TempDir File tempDir) throws Exception { + File sourceSavepointDir = new File(tempDir, "sp-old"); + File targetSavepointDir = new File(tempDir, "sp-new"); + assertThat(sourceSavepointDir.mkdirs()).isTrue(); + assertThat(targetSavepointDir.mkdirs()).isTrue(); + + // A retained state file living in the source savepoint, referenced by a relative handle + // whose full path still points at the file in the source directory (this is what loading + // an existing savepoint produces). + byte[] stateBytes = "retained-shared-state".getBytes(StandardCharsets.UTF_8); + File sourceStateFile = new File(sourceSavepointDir, STATE_FILE_NAME); + Files.write(sourceStateFile.toPath(), stateBytes); + + Path targetSavepointPath = new Path(targetSavepointDir.getAbsolutePath()); + Path sourceStatePath = + new Path(new Path(sourceSavepointDir.getAbsolutePath()), STATE_FILE_NAME); + RelativeFileStateHandle retainedHandle = + new RelativeFileStateHandle(sourceStatePath, STATE_FILE_NAME, stateBytes.length); + + // Copy the retained file the way SavepointWriter.fromExistingSavepoint does: through + // FileCopyFunction, which copies by bare file name into the new savepoint directory. The + // metadata written below is self-contained only together with this naming contract. + FileCopyFunction copyFunction = new FileCopyFunction(targetSavepointDir.getAbsolutePath()); + copyFunction.open(FirstAttemptInitializationContext.of(SUBTASK_INDEX, PARALLELISM)); + copyFunction.writeRecord(sourceStatePath); + copyFunction.close(); + + CheckpointMetadata metadata = createMetadata(retainedHandle); + + SavepointOutputFormat format = new SavepointOutputFormat(targetSavepointPath); + format.setRuntimeContext(new MockStreamingRuntimeContext(PARALLELISM, SUBTASK_INDEX)); + format.open(FirstAttemptInitializationContext.of(SUBTASK_INDEX, PARALLELISM)); + format.writeRecord(metadata); + format.close(); + + // Reload the written metadata; relative handles resolve against the directory the + // metadata is read from, i.e. the new savepoint holding the copies. + CheckpointMetadata reloaded = + SavepointLoader.loadSavepointMetadata(targetSavepointPath.getPath()); + StreamStateHandle reloadedShared = extractSingleSharedStateHandle(reloaded); + + assertThat(reloadedShared) + .as( + "Retained state must reload as a relative handle that resolves inside " + + "the new savepoint; an absolute path back into the source " + + "savepoint would break the new savepoint once the source is " + + "deleted") + .isInstanceOf(RelativeFileStateHandle.class); + + RelativeFileStateHandle reloadedRelative = (RelativeFileStateHandle) reloadedShared; + assertThat(reloadedRelative.getRelativePath()).isEqualTo(STATE_FILE_NAME); + assertThat(reloadedRelative.getFilePath().getPath()) + .isEqualTo(new Path(targetSavepointPath, STATE_FILE_NAME).getPath()); + assertThat(new File(reloadedRelative.getFilePath().getPath())).exists(); + } + + private static CheckpointMetadata createMetadata(RelativeFileStateHandle retainedHandle) { + StreamStateHandle metaStateHandle = + new ByteStreamStateHandle( + "backend-meta", "backend-meta".getBytes(StandardCharsets.UTF_8)); + IncrementalRemoteKeyedStateHandle keyedStateHandle = + new IncrementalRemoteKeyedStateHandle( + UUID.randomUUID(), + KeyGroupRange.of(0, MAX_PARALLELISM - 1), + CHECKPOINT_ID, + Collections.singletonList( + HandleAndLocalPath.of(retainedHandle, STATE_FILE_NAME)), + Collections.emptyList(), + metaStateHandle); + + OperatorSubtaskState subtaskState = + OperatorSubtaskState.builder().setManagedKeyedState(keyedStateHandle).build(); + + OperatorState operatorState = + new OperatorState( + null, + OPERATOR_UID, + OperatorIDGenerator.fromUid(OPERATOR_UID), + PARALLELISM, + MAX_PARALLELISM); + operatorState.putState(SUBTASK_INDEX, subtaskState); + + return new CheckpointMetadata( + CHECKPOINT_ID, Collections.singleton(operatorState), Collections.emptyList()); + } + + private static StreamStateHandle extractSingleSharedStateHandle(CheckpointMetadata metadata) { + IncrementalRemoteKeyedStateHandle keyedStateHandle = + (IncrementalRemoteKeyedStateHandle) + metadata.getOperatorStates() + .iterator() + .next() + .getState(SUBTASK_INDEX) + .getManagedKeyedState() + .iterator() + .next(); + return keyedStateHandle.getSharedState().get(0).getHandle(); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java index 4f1000fd3b370..b167ed401f1f8 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,12 +29,15 @@ 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.CheckpointedStateScope; import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation; import org.apache.flink.runtime.state.StateBackend; import org.apache.flink.runtime.state.StateBackendLoader; import org.apache.flink.runtime.state.StreamStateHandle; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage; import org.apache.flink.util.CollectionUtil; @@ -64,6 +68,14 @@ * 

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

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

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

The deliberately different method name (rather than an overload) makes choosing this + * encoding an explicit decision at the call site. + */ + public static void storeCheckpointMetadataWithoutExclusiveDir( CheckpointMetadata checkpointMetadata, OutputStream out) throws IOException { DataOutputStream dos = new DataOutputStream(out); - storeCheckpointMetadata(checkpointMetadata, dos); + storeCheckpointMetadataWithoutExclusiveDir(checkpointMetadata, dos); } + /** + * Stores the checkpoint metadata, letting the serializer check every {@link + * RelativeFileStateHandle} against the exclusive directory the stream writes into ({@link + * CheckpointMetadataOutputStream#getExclusiveCheckpointDir()}): a handle whose file lives in + * that directory keeps the relative encoding, while a handle pointing anywhere else (e.g. an + * SST file reused from a claimed savepoint) is persisted with its absolute path, because the + * relative form would be re-anchored to the wrong directory when the metadata is read back. + * Use this variant when writing an actual checkpoint or savepoint. + */ public static void storeCheckpointMetadata( + CheckpointMetadata checkpointMetadata, CheckpointMetadataOutputStream out) + 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/ChannelStateHandleSerializerV1.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java index 22eefc63cf06d..75f52e933d333 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV1.java @@ -99,7 +99,9 @@ static void serializeChannelStateHandle( dos.writeLong(offset); } dos.writeLong(handle.getStateSize()); - serializeStreamStateHandle(handle.getDelegate(), dos); + // null context: channel state is always written into the current checkpoint's own + // exclusive directory, see MetadataV3Serializer#serializeSubtaskState. + serializeStreamStateHandle(handle.getDelegate(), dos, null); } static > diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java index ba8f8ea6c6ce2..0bd9f693e47ff 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializerV2.java @@ -146,7 +146,9 @@ private void serializeMergedChannelStateHandle( dos.writeInt(handle.getSubtaskIndex()); dos.writeLong(handle.getStateSize()); - serializeStreamStateHandle(handle.getDelegate(), dos); + // null context: channel state is always written into the current checkpoint's own + // exclusive directory, see MetadataV3Serializer#serializeSubtaskState. + serializeStreamStateHandle(handle.getDelegate(), dos, null); dos.writeInt(handle.getSerializedChannelOffsets().length); dos.write(handle.getSerializedChannelOffsets()); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java index 543b2c5fc5b07..c80108ca2d362 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,7 +18,11 @@ package org.apache.flink.runtime.checkpoint.metadata; +import org.apache.flink.core.fs.Path; import org.apache.flink.core.io.Versioned; +import org.apache.flink.runtime.checkpoint.Checkpoints; + +import javax.annotation.Nullable; import java.io.DataInputStream; import java.io.DataOutputStream; @@ -45,10 +49,42 @@ CheckpointMetadata deserialize( DataInputStream dis, ClassLoader userCodeClassLoader, String externalPointer) throws IOException; + /** + * Convenience variant of {@link #serialize(CheckpointMetadata, DataOutputStream, Path)} with a + * {@code null} exclusive directory: every relative file reference is kept relative, + * unconditionally, and is resolved against whatever directory the metadata is later read from. + * Only correct if every referenced file ends up in that directory; see {@link + * Checkpoints#storeCheckpointMetadataWithoutExclusiveDir} for the use case. + * + * @throws IOException Serialization failures are forwarded + */ + default void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + throws IOException { + serialize(checkpointMetadata, dos, null); + } + /** * Serializes a savepoint or checkpoint metadata to an output stream. * + *

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

This is the method implementations must provide (rather than a default dropping the + * exclusive directory), so that a serializer cannot silently lose the exclusive-directory + * information. + * + * @param exclusiveDirPath the directory that will contain the metadata file (the checkpoint's + * exclusive directory), or {@code null} to keep every relative reference relative + * regardless of where its file lives * @throws IOException Serialization failures are forwarded */ - void serialize(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 d1e0075f7ae0f..8d35eb72c4d5b 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 c993acb1c4d4a..8f324864167db 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 57d69b29f3e56..6825418f4b84b 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( @@ -323,7 +338,10 @@ protected OperatorSubtaskState deserializeSubtaskState( // ------------------------------------------------------------------------ @VisibleForTesting - static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputStream dos) + static void serializeKeyedStateHandle( + KeyedStateHandle stateHandle, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { if (stateHandle == null) { dos.writeByte(NULL_HANDLE); @@ -340,7 +358,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 +376,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 +396,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 +431,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 +617,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 +658,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); } @@ -716,14 +740,23 @@ protected void serializeInputStateHandle(InputStateHandle handle, DataOutputStre // low-level state handles // ------------------------------------------------------------------------ - static void serializeStreamStateHandle(StreamStateHandle stateHandle, DataOutputStream dos) + static void serializeStreamStateHandle( + StreamStateHandle stateHandle, + DataOutputStream dos, + @Nullable SerializationContext context) throws IOException { if (stateHandle == null) { dos.writeByte(NULL_HANDLE); - } else if (stateHandle instanceof RelativeFileStateHandle) { - dos.writeByte(RELATIVE_STREAM_STATE_HANDLE); + } else if (stateHandle instanceof RelativeFileStateHandle + && canKeepRelativeEncoding((RelativeFileStateHandle) stateHandle, context)) { + // The relative encoding stores just the file name; on restore it is resolved + // against the directory the metadata is read from. So it is only used for files + // that live in the directory being written. Handles to files anywhere else fail + // canKeepRelativeEncoding and take the FileStateHandle branch below, which keeps + // their full absolute path. RelativeFileStateHandle relativeFileStateHandle = (RelativeFileStateHandle) stateHandle; + dos.writeByte(RELATIVE_STREAM_STATE_HANDLE); dos.writeUTF(relativeFileStateHandle.getRelativePath()); dos.writeLong(relativeFileStateHandle.getStateSize()); } else if (stateHandle instanceof SegmentFileStateHandle) { @@ -760,13 +793,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 +854,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 +945,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 +976,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 +1014,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 012914ab14da1..261b668ea9b57 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( @@ -248,6 +263,10 @@ private void serializeCollection( // ------------------------------------------------------------------------ // exposed static methods for test cases // + // These helpers (de)serialize standalone handles outside any checkpoint, + // so there is no exclusive directory to check relative references + // against: the (de)serialization context is always null. + // // NOTE: The fact that certain tests directly call these lower level // serialization methods is a problem, because that way the tests // bypass the versioning scheme. Especially tests that test for @@ -258,7 +277,7 @@ private void serializeCollection( @VisibleForTesting public static void serializeStreamStateHandle( StreamStateHandle stateHandle, DataOutputStream dos) throws IOException { - MetadataV2V3SerializerBase.serializeStreamStateHandle(stateHandle, dos); + MetadataV2V3SerializerBase.serializeStreamStateHandle(stateHandle, dos, null); } @VisibleForTesting @@ -270,7 +289,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 +301,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 88262fa805a47..f49be1ce902a5 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 ca0d568ccf4b4..8dd17d1b9ef5e 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 525e21b0b5eda..d4a4a4170a7fa 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 4f5c2596ba02f..ef6a1120b4ce7 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 212930044fa73..16cf4b3370510 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 834c73ec4b250..5c73e68acf8eb 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 0000000000000..7989b187e21bc --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java @@ -0,0 +1,183 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.checkpoint.metadata; + +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.CheckpointProperties; +import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath; +import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle; +import org.apache.flink.runtime.state.memory.ByteStreamStateHandle; + +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests that the {@link MetadataSerializer#serialize} entry points agree with each other across + * every registered metadata version: + * + *

+ * + *

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

A {@link RelativeFileStateHandle} only stores the file name relative to the exclusive + * directory it was written into. On recovery the absolute path is rebuilt as {@code new + * Path(, relativePath)}. This is only correct as long as the file physically lives in + * that exclusive directory. The serializer therefore keeps a relative handle relative only when it + * belongs to the exclusive directory being written, and otherwise persists its absolute path. + * + *

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

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

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

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

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

{@code RocksDBClaimSavepointThenCheckpointRestoreTest} pins the same property at the + * backend/serializer level, in-process, without the JobManager finalization wiring. This test + * instead drives the whole path on a real cluster, so it also covers what the component test + * cannot: in particular that the metadata finalization in {@code PendingCheckpoint} binds to the + * {@code CheckpointMetadataOutputStream} overload of {@code Checkpoints#storeCheckpointMetadata}, + * which hands the checkpoint's exclusive directory to the serializer. + */ +class ResumeCheckpointAfterClaimedNativeSavepointITCase { + + private static final int PARALLELISM = 2; + private static final int NUM_KEYS = 128; + + @TempDir private Path checkpointsDir; + @TempDir private Path savepointsDir; + + @Test + void testFirstCheckpointAfterClaimIsRestorable() throws Exception { + final Configuration config = new Configuration(); + config.set(StateBackendOptions.STATE_BACKEND, "rocksdb"); + config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true); + config.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, false); + config.set(CheckpointingOptions.FILE_MERGING_ENABLED, false); + config.set(RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE, false); + config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointsDir.toUri().toString()); + config.set( + CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION, + ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION); + // Keep every state file a real file so the shared-file assertion sees actual paths. + config.set(CheckpointingOptions.FS_SMALL_FILE_THRESHOLD, MemorySize.ZERO); + // Fail fast on a broken restore instead of looping through restarts. + config.set(RestartStrategyOptions.RESTART_STRATEGY, "none"); + + // Manual lifecycle because the configuration depends on per-test @TempDirs. + final MiniClusterWithClientResource cluster = + new MiniClusterWithClientResource( + new MiniClusterResourceConfiguration.Builder() + .setConfiguration(config) + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(PARALLELISM) + .build()); + cluster.before(); + try { + final ClusterClient client = cluster.getClusterClient(); + final MiniCluster miniCluster = cluster.getMiniCluster(); + + // Checkpoint once so the savepoint covers a mix of reused and freshly flushed SSTs, + // then stop with a NATIVE savepoint. + final JobGraph initialJob = createJobGraph(config); + client.submitJob(initialJob).get(); + waitForAllTaskRunning(miniCluster, initialJob.getJobID(), false); + miniCluster.triggerCheckpoint(initialJob.getJobID()).get(); + final String savepointPath = + client.stopWithSavepoint( + initialJob.getJobID(), + false, + savepointsDir.toUri().toString(), + SavepointFormatType.NATIVE) + .get(); + + // Restore in CLAIM mode, the first incremental checkpoint reuses the savepoint's SSTs. + final JobGraph restoredFromSavepoint = createJobGraph(config); + restoredFromSavepoint.setSavepointRestoreSettings( + SavepointRestoreSettings.forPath( + savepointPath, false, RecoveryClaimMode.CLAIM)); + client.submitJob(restoredFromSavepoint).get(); + waitForAllTaskRunning(miniCluster, restoredFromSavepoint.getJobID(), false); + final String checkpointPath = + miniCluster.triggerCheckpoint(restoredFromSavepoint.getJobID()).get(); + client.cancel(restoredFromSavepoint.getJobID()).get(); + miniCluster.requestJobResult(restoredFromSavepoint.getJobID()).get(); + + // The regression guard: reused savepoint files must be recorded by their absolute + // location. The trailing separator keeps a sibling directory prefix from matching. + final String savepointPrefix = + savepointPath.endsWith("/") ? savepointPath : savepointPath + "/"; + assertThat(sharedStateFilePaths(checkpointPath)) + .as( + "the first incremental checkpoint after claiming the savepoint must " + + "reference reused savepoint files by their absolute location") + .isNotEmpty() + .anySatisfy(path -> assertThat(path).startsWith(savepointPrefix)); + + // Restore from that checkpoint (a missing-file failure would go terminal and surface + // here) and let the restored job complete one more checkpoint of its own. + final JobGraph restoredFromCheckpoint = createJobGraph(config); + restoredFromCheckpoint.setSavepointRestoreSettings( + SavepointRestoreSettings.forPath(checkpointPath, false)); + client.submitJob(restoredFromCheckpoint).get(); + waitForAllTaskRunning(miniCluster, restoredFromCheckpoint.getJobID(), false); + miniCluster.triggerCheckpoint(restoredFromCheckpoint.getJobID()).get(); + client.cancel(restoredFromCheckpoint.getJobID()).get(); + miniCluster.requestJobResult(restoredFromCheckpoint.getJobID()).get(); + } finally { + cluster.after(); + } + } + + /** All absolute file paths referenced by the checkpoint's keyed shared state. */ + private static List sharedStateFilePaths(String checkpointPath) throws Exception { + final CheckpointMetadata metadata = loadCheckpointMetadata(checkpointPath); + return metadata.getOperatorStates().stream() + .flatMap(operatorState -> operatorState.getStates().stream()) + .flatMap(subtaskState -> subtaskState.getManagedKeyedState().stream()) + .filter(IncrementalRemoteKeyedStateHandle.class::isInstance) + .map(IncrementalRemoteKeyedStateHandle.class::cast) + .flatMap(handle -> handle.getSharedState().stream()) + .map(HandleAndLocalPath::getHandle) + .filter(FileStateHandle.class::isInstance) + .map(handle -> ((FileStateHandle) handle).getFilePath().toString()) + .collect(Collectors.toList()); + } + + private static JobGraph createJobGraph(Configuration config) { + final StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(config); + env.setParallelism(PARALLELISM); + + // The throttle is chained to the source, so barriers never queue behind a record backlog. + // Not the full Long range: splitting that overflows in NumberSequenceIterator#split. + env.fromSequence(0, Long.MAX_VALUE - 1) + .map(new Throttler()) + .keyBy(value -> Math.floorMod(value, NUM_KEYS)) + .map(new StatefulCounter()) + .uid("stateful-counter") + .sinkTo(new DiscardingSink<>()); + + return env.getStreamGraph().getJobGraph(); + } + + /** + * Slows production so the reused savepoint SSTs are not compacted away before the post-restore + * checkpoint. Best-effort: if reuse stops happening, the shared-file assertion fails. + */ + private static final class Throttler implements MapFunction { + + private static final long THROTTLE_MILLIS = 1; + + @Override + public Long map(Long value) throws Exception { + Thread.sleep(THROTTLE_MILLIS); + return value; + } + } + + /** Gives the job keyed RocksDB state, without it there would be no shared SSTs to assert on. */ + private static final class StatefulCounter extends RichMapFunction { + + private ValueState counter; + + @Override + public void open(OpenContext openContext) throws Exception { + counter = + getRuntimeContext() + .getState( + new ValueStateDescriptor<>( + "counter", BasicTypeInfo.LONG_TYPE_INFO)); + } + + @Override + public Long map(Long value) throws Exception { + long next = Optional.ofNullable(counter.value()).orElse(0L) + value; + counter.update(next); + return next; + } + } +}