This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5bf06e59bab28f28ab6e828192bb7ea42f95053e Author: Aleksandr Savonin <[email protected]> AuthorDate: Wed Jul 29 10:05:09 2026 +0200 [FLINK-39964][state] Fix unrestorable checkpoint after CLAIM restore from native savepoint The first incremental checkpoint after a CLAIM-mode restore from a native savepoint reuses the savepoint's SST files as RelativeFileStateHandles. The metadata serializer stored only their relative path, so on restore they were resolved against the new checkpoint's exclusive directory and could no longer be found. The serializer now knows the exclusive directory of the checkpoint being written (CheckpointMetadataOutputStream#getExclusiveCheckpointDir, passed down as a non-null SerializationContext) and keeps the compact relative encoding only for handles whose file actually lives in that directory; a foreign handle (a reused savepoint SST) is persisted with its absolute path using the pre-existing FileStateHandle encoding. The wire format and the metadata version are unchanged. The state processor API keeps the unconditional relative encoding via the explicit Checkpoints#storeCheckpointMetadataWithoutExclusiveDir, because its deep copy places every referenced file next to the new metadata. Generated-by: Claude Opus 4.8 (1M context) (cherry picked from commit 07bcbd9f1bef8329e68d4837e16a5a963b2818bc) --- .../state/api/output/SavepointOutputFormat.java | 15 +- .../flink/state/api/SavepointDeepCopyTest.java | 6 + .../SavepointOutputFormatSelfContainedTest.java | 175 ++++++++ .../flink/runtime/checkpoint/Checkpoints.java | 76 +++- .../metadata/ChannelStateHandleSerializerV1.java | 7 +- .../metadata/ChannelStateHandleSerializerV2.java | 7 +- .../checkpoint/metadata/MetadataSerializer.java | 20 +- .../checkpoint/metadata/MetadataV1Serializer.java | 8 +- .../checkpoint/metadata/MetadataV2Serializer.java | 16 +- .../metadata/MetadataV2V3SerializerBase.java | 155 +++++-- .../checkpoint/metadata/MetadataV3Serializer.java | 42 +- .../checkpoint/metadata/MetadataV4Serializer.java | 10 +- .../checkpoint/metadata/MetadataV5Serializer.java | 5 +- .../state/CheckpointMetadataOutputStream.java | 26 +- .../FsCheckpointMetadataOutputStream.java | 5 + .../checkpoint/CheckpointMetadataLoadingTest.java | 2 +- .../flink/runtime/checkpoint/CheckpointsTest.java | 3 +- .../MetadataSerializerEntryPointsTest.java | 155 +++++++ .../metadata/MetadataV2V3SerializerBaseTest.java | 443 +++++++++++++++++++++ .../metadata/MetadataV3SerializerTest.java | 10 +- .../metadata/MetadataV4SerializerTest.java | 2 +- .../metadata/MetadataV5SerializerTest.java | 4 +- .../metadata/MetadataV6SerializerTest.java | 2 +- .../apache/flink/runtime/jobmaster/TestUtils.java | 2 +- .../OperatorCoordinatorSchedulerTest.java | 2 +- ...sDBClaimSavepointThenCheckpointRestoreTest.java | 350 ++++++++++++++++ ...heckpointAfterClaimedNativeSavepointITCase.java | 280 +++++++++++++ 27 files changed, 1760 insertions(+), 68 deletions(-) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java index c0d022e04a7..dd8a89b655e 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 class SavepointOutputFormat extends RichOutputFormat<CheckpointMetadata> () -> { try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) { - Checkpoints.storeCheckpointMetadata(metadata, out); + // Must use the WITHOUT-exclusive-directory method. + // The State Processor API guarantees that + // every file referenced by this _metadata physically lives in + // the new savepoint folder: + // 1. FileCopyFunction copies carried-over operators' files into it. + // 2. New operators write their files directly into it. + // So relative, filename-only encoding resolves to a local file on + // restore, and the savepoint stays self-contained. + // The exclusive-dir-aware Checkpoints.storeCheckpointMetadata + // would instead persist carried-over handles' absolute paths + // that point to the source savepoint, + // breaking this savepoint once the source is deleted. + Checkpoints.storeCheckpointMetadataWithoutExclusiveDir( + metadata, out); CompletedCheckpointStorageLocation finalizedLocation = out.closeAndFinalizeCheckpoint(); return finalizedLocation.getExternalPointer(); diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java index a633c59dd05..56f3c6f044f 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.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.test.util.AbstractTestBaseJUnit4; import org.apache.flink.util.AbstractID; import org.apache.flink.util.Collector; +import org.apache.flink.util.FileUtils; import org.apache.commons.lang3.RandomStringUtils; import org.junit.Assert; @@ -184,6 +185,11 @@ public class SavepointDeepCopyTest extends AbstractTestBaseJUnit4 { 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 00000000000..003c45e575b --- /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. + * + * <p>{@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. + * + * <p>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 4f1000fd3b3..1579cc0ef6e 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 @@ package org.apache.flink.runtime.checkpoint; 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.checkpoint.metadata.MetadataV6Serializer; 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 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * <pre>[MagicNumber (int) | Format Version (int) | Checkpoint Metadata (variable)]</pre> * * <p>The actual savepoint serialization is version-specific via the {@link MetadataSerializer}. + * + * <p>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 + * <i>exclusive directory</i>: the per-checkpoint directory that contains the {@code _metadata} file + * and the checkpoint's own, non-shared state files, e.g. {@code /checkpoints/<job-id>/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)}. + * + * <p>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 22eefc63cf0..fe814b27313 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,12 @@ class ChannelStateHandleSerializerV1 implements ChannelStateHandleSerializer { dos.writeLong(offset); } dos.writeLong(handle.getStateSize()); - serializeStreamStateHandle(handle.getDelegate(), dos); + // No exclusive directory: channel state is always written into the current checkpoint's + // own exclusive directory, see MetadataV3Serializer#serializeSubtaskState. + serializeStreamStateHandle( + handle.getDelegate(), + dos, + MetadataV2V3SerializerBase.SerializationContext.withoutExclusiveDir()); } static <Info, Handle extends AbstractChannelStateHandle<Info>> 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 ba8f8ea6c6c..40304f7431a 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,12 @@ class ChannelStateHandleSerializerV2 implements ChannelStateHandleSerializer { dos.writeInt(handle.getSubtaskIndex()); dos.writeLong(handle.getStateSize()); - serializeStreamStateHandle(handle.getDelegate(), dos); + // No exclusive directory: channel state is always written into the current checkpoint's + // own exclusive directory, see MetadataV3Serializer#serializeSubtaskState. + serializeStreamStateHandle( + handle.getDelegate(), + dos, + MetadataV2V3SerializerBase.SerializationContext.withoutExclusiveDir()); 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 543b2c5fc5b..0abafe0052a 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; @@ -48,7 +52,21 @@ public interface MetadataSerializer extends Versioned { /** * Serializes a savepoint or checkpoint metadata to an output stream. * + * <p>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. + * + * @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 d1e0075f7ae..8d35eb72c4d 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 class MetadataV1Serializer implements MetadataSerializer { } @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 c993acb1c4d..0fbd13d4c7d 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 class MetadataV2Serializer extends MetadataV2V3SerializerBase implements } @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,8 @@ public class MetadataV2Serializer extends MetadataV2V3SerializerBase implements // ------------------------------------------------------------------------ @Override - protected void serializeOperatorState(OperatorState operatorState, DataOutputStream dos) + protected void serializeOperatorState( + OperatorState operatorState, DataOutputStream dos, SerializationContext context) throws IOException { checkState( !operatorState.isFullyFinished(), @@ -102,7 +107,7 @@ public class MetadataV2Serializer extends MetadataV2V3SerializerBase implements dos.writeInt(subtaskStateMap.size()); for (Map.Entry<Integer, OperatorSubtaskState> entry : subtaskStateMap.entrySet()) { dos.writeInt(entry.getKey()); - serializeSubtaskState(entry.getValue(), dos); + serializeSubtaskState(entry.getValue(), dos, context); } } @@ -134,7 +139,8 @@ public class MetadataV2Serializer extends MetadataV2V3SerializerBase implements } @Override - protected void serializeSubtaskState(OperatorSubtaskState subtaskState, DataOutputStream dos) + protected void serializeSubtaskState( + OperatorSubtaskState subtaskState, DataOutputStream dos, SerializationContext context) throws IOException { // write two unused fields for compatibility: // - "duration" @@ -142,7 +148,7 @@ public class MetadataV2Serializer extends MetadataV2V3SerializerBase implements 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 57d69b29f3e..1ad9768312e 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,8 +145,12 @@ public abstract class MetadataV2V3SerializerBase { // (De)serialization entry points // ------------------------------------------------------------------------ - protected void serializeMetadata(CheckpointMetadata checkpointMetadata, DataOutputStream dos) + protected void serializeMetadata( + CheckpointMetadata checkpointMetadata, + DataOutputStream dos, + SerializationContext context) throws IOException { + Preconditions.checkNotNull(context, "SerializationContext"); // first: checkpoint ID dos.writeLong(checkpointMetadata.getCheckpointId()); @@ -162,7 +166,7 @@ public abstract class MetadataV2V3SerializerBase { dos.writeInt(operatorStates.size()); for (OperatorState operatorState : operatorStates) { - serializeOperatorState(operatorState, dos); + serializeOperatorState(operatorState, dos, context); } } @@ -267,25 +271,33 @@ public abstract class MetadataV2V3SerializerBase { // ------------------------------------------------------------------------ protected abstract void serializeOperatorState( - OperatorState operatorState, DataOutputStream dos) throws IOException; + OperatorState operatorState, DataOutputStream dos, 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, 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<KeyedStateHandle> managedKeyedState, DataOutputStream dos) + StateObjectCollection<KeyedStateHandle> managedKeyedState, + DataOutputStream dos, + SerializationContext context) throws IOException { - serializeKeyedStateHandle(extractSingleton(managedKeyedState), dos); + serializeKeyedStateHandle(extractSingleton(managedKeyedState), dos, context); } protected OperatorSubtaskState deserializeSubtaskState( @@ -323,7 +335,8 @@ public abstract class MetadataV2V3SerializerBase { // ------------------------------------------------------------------------ @VisibleForTesting - static void serializeKeyedStateHandle(KeyedStateHandle stateHandle, DataOutputStream dos) + static void serializeKeyedStateHandle( + KeyedStateHandle stateHandle, DataOutputStream dos, SerializationContext context) throws IOException { if (stateHandle == null) { dos.writeByte(NULL_HANDLE); @@ -340,7 +353,7 @@ public abstract class MetadataV2V3SerializerBase { 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 +371,13 @@ public abstract class MetadataV2V3SerializerBase { 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 +391,12 @@ public abstract class MetadataV2V3SerializerBase { 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 +426,7 @@ public abstract class MetadataV2V3SerializerBase { for (Tuple2<StreamStateHandle, Long> 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 +612,8 @@ public abstract class MetadataV2V3SerializerBase { stateHandleId); } - void serializeOperatorStateHandle(OperatorStateHandle stateHandle, DataOutputStream dos) + void serializeOperatorStateHandle( + OperatorStateHandle stateHandle, DataOutputStream dos, SerializationContext context) throws IOException { if (stateHandle != null) { dos.writeByte( @@ -634,7 +651,7 @@ public abstract class MetadataV2V3SerializerBase { .toString()); dos.writeBoolean(stateHandle instanceof EmptyFileMergingOperatorStreamStateHandle); } - serializeStreamStateHandle(stateHandle.getDelegateStateHandle(), dos); + serializeStreamStateHandle(stateHandle.getDelegateStateHandle(), dos, context); } else { dos.writeByte(NULL_HANDLE); } @@ -716,14 +733,18 @@ public abstract class MetadataV2V3SerializerBase { // low-level state handles // ------------------------------------------------------------------------ - static void serializeStreamStateHandle(StreamStateHandle stateHandle, DataOutputStream dos) + static void serializeStreamStateHandle( + StreamStateHandle stateHandle, DataOutputStream dos, 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)) { + // Foreign handles fail canKeepRelativeEncoding and fall through to 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 +781,45 @@ public abstract class MetadataV2V3SerializerBase { 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} keeps the relative encoding or must be + * written with its absolute path like a plain {@link FileStateHandle}. + * + * <p>A relative reference is rebuilt on read as {@code <directory containing the + * metadata>/<relative path>} (see {@link #resolveRelativeHandle}), so a handle may only stay + * relative if its file actually lives in the exclusive directory being written. A handle + * pointing anywhere else (e.g. a claimed savepoint's SST that a later incremental checkpoint + * still references) would be resolved to a path where its file never existed, so it keeps its + * absolute path. + * + * <p>Without an exclusive directory ({@link SerializationContext#withoutExclusiveDir()}) every + * relative reference stays relative unconditionally. + */ + private static boolean canKeepRelativeEncoding( + RelativeFileStateHandle handle, SerializationContext context) { + Path exclusiveDirPath = 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 +844,7 @@ public abstract class MetadataV2V3SerializerBase { } 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 +935,13 @@ public abstract class MetadataV2V3SerializerBase { } private static void serializeHandleAndLocalPathList( - List<HandleAndLocalPath> list, DataOutputStream dos) throws IOException { + List<HandleAndLocalPath> list, DataOutputStream dos, 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 +964,51 @@ public abstract class MetadataV2V3SerializerBase { // internal helper classes // ------------------------------------------------------------------------ + /** + * A context that keeps information needed while <i>serializing</i> checkpoint metadata. It is + * the write-side counterpart of {@link DeserializationContext} and is passed through the + * serialize methods in the same way. + * + * <p>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 { + + private static final SerializationContext WITHOUT_EXCLUSIVE_DIR = + new SerializationContext(null); + + @Nullable private final Path exclusiveDirPath; + + private SerializationContext(@Nullable Path exclusiveDirPath) { + this.exclusiveDirPath = exclusiveDirPath; + } + + /** + * Context for metadata written into a known exclusive directory: a {@link + * RelativeFileStateHandle} keeps its relative encoding only when its file lives in {@code + * exclusiveDirPath}, otherwise it is persisted with its absolute path. + */ + static SerializationContext withExclusiveDir(Path exclusiveDirPath) { + return new SerializationContext(Preconditions.checkNotNull(exclusiveDirPath)); + } + + /** + * Context for metadata written without a known exclusive directory, so every {@link + * RelativeFileStateHandle} keeps its relative encoding unconditionally. Correct only when + * every referenced file ends up next to the metadata; each call site documents why that + * holds for it. + */ + static SerializationContext withoutExclusiveDir() { + return WITHOUT_EXCLUSIVE_DIR; + } + + @Nullable + Path getExclusiveDirPath() { + return exclusiveDirPath; + } + } + /** * A context that keeps information needed during serialization. This context is passed along by * the methods. In some sense, this replaces the member fields of the class, because the @@ -925,7 +1024,7 @@ public abstract class MetadataV2V3SerializerBase { * <p>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 012914ab14d..79c67ef49d3 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 @@ package org.apache.flink.runtime.checkpoint.metadata; 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,17 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements // ------------------------------------------------------------------------ @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, + exclusiveDirPath == null + ? SerializationContext.withoutExclusiveDir() + : SerializationContext.withExclusiveDir(exclusiveDirPath)); } @Override @@ -95,7 +104,8 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements // ------------------------------------------------------------------------ @Override - protected void serializeOperatorState(OperatorState operatorState, DataOutputStream dos) + protected void serializeOperatorState( + OperatorState operatorState, DataOutputStream dos, SerializationContext context) throws IOException { // Operator ID dos.writeLong(operatorState.getOperatorID().getLowerPart()); @@ -106,7 +116,7 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements dos.writeInt(operatorState.getMaxParallelism()); // Coordinator state - serializeStreamStateHandle(operatorState.getCoordinatorState(), dos); + serializeStreamStateHandle(operatorState.getCoordinatorState(), dos, context); // Sub task states if (operatorState.isFullyFinished()) { @@ -119,7 +129,7 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements boolean isFinished = entry.getValue().isFinished(); serializeSubtaskIndexAndFinishedState(entry.getKey(), isFinished, dos); if (!isFinished) { - serializeSubtaskState(entry.getValue(), dos); + serializeSubtaskState(entry.getValue(), dos, context); } } } @@ -137,9 +147,15 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements } @Override - protected void serializeSubtaskState(OperatorSubtaskState subtaskState, DataOutputStream dos) + protected void serializeSubtaskState( + OperatorSubtaskState subtaskState, DataOutputStream dos, 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 +264,11 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements // ------------------------------------------------------------------------ // 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: they serialize with SerializationContext.withoutExclusiveDir() + // and deserialize with a null DeserializationContext. + // // 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 +279,8 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements @VisibleForTesting public static void serializeStreamStateHandle( StreamStateHandle stateHandle, DataOutputStream dos) throws IOException { - MetadataV2V3SerializerBase.serializeStreamStateHandle(stateHandle, dos); + MetadataV2V3SerializerBase.serializeStreamStateHandle( + stateHandle, dos, SerializationContext.withoutExclusiveDir()); } @VisibleForTesting @@ -270,7 +292,7 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements @VisibleForTesting public void serializeOperatorStateHandleUtil( OperatorStateHandle stateHandle, DataOutputStream dos) throws IOException { - serializeOperatorStateHandle(stateHandle, dos); + serializeOperatorStateHandle(stateHandle, dos, SerializationContext.withoutExclusiveDir()); } @VisibleForTesting @@ -282,7 +304,7 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements @VisibleForTesting public void serializeKeyedStateHandleUtil(KeyedStateHandle stateHandle, DataOutputStream dos) throws IOException { - serializeKeyedStateHandle(stateHandle, dos); + serializeKeyedStateHandle(stateHandle, dos, SerializationContext.withoutExclusiveDir()); } @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 88262fa805a..f49be1ce902 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 class MetadataV4Serializer extends MetadataV3Serializer { } @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 ca0d568ccf4..ce06b65c2e7 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,8 @@ public class MetadataV5Serializer extends MetadataV4Serializer { } @Override - protected void serializeOperatorState(OperatorState operatorState, DataOutputStream dos) + protected void serializeOperatorState( + OperatorState operatorState, DataOutputStream dos, SerializationContext context) throws IOException { if (operatorState.getOperatorName().isPresent() && operatorState.getOperatorName().get().isEmpty()) { @@ -60,7 +61,7 @@ public class MetadataV5Serializer extends MetadataV4Serializer { // 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 525e21b0b5e..3f56918b5dd 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 @@ import java.io.IOException; * An output stream for checkpoint metadata. * * <p>This stream is similar to the {@link CheckpointStateOutputStream}, but for metadata files - * rather thancdata files. + * rather than data files. * * <p>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). + * + * <p>This is used while serializing the metadata to keep relative file references + * self-consistent (see {@code MetadataV2V3SerializerBase}). + * + * <p>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} keeps every relative + * handle relative unconditionally, 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 4f5c2596ba0..ef6a1120b4c 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 final class FsCheckpointMetadataOutputStream extends CheckpointMetadataOu 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 212930044fa..16cf4b33705 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 @@ class CheckpointMetadataLoadingTest { 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 834c73ec4b2..5c73e68acf8 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 @@ class CheckpointsTest { 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 00000000000..96b6f6192f6 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializerEntryPointsTest.java @@ -0,0 +1,155 @@ +/* + * 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 single {@link MetadataSerializer#serialize} entry point behaves correctly across + * every registered metadata version: + * + * <ul> + * <li>a supplied exclusive directory reaches the relative-handle encoder instead of being + * silently dropped: making a shared relative handle foreign to it changes the written bytes; + * <li>the deserialize-only versions reject serialization. + * </ul> + * + * <p>What the exclusive directory means 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 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 testDeserializeOnlyVersionsRejectSerialization() { + final CheckpointMetadata metadata = metadataWithForeignSharedState(); + for (MetadataSerializer serializer : deserializeOnlySerializers()) { + assertThatThrownBy(() -> serialize(serializer, metadata, null)) + .as("serialize must be unsupported for v%d", serializer.getVersion()) + .isInstanceOf(UnsupportedOperationException.class); + } + } + + // ------------------------------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------------------------------ + + private static List<MetadataSerializer> writableSerializers() { + return Arrays.asList( + MetadataV3Serializer.INSTANCE, + MetadataV4Serializer.INSTANCE, + MetadataV5Serializer.INSTANCE, + MetadataV6Serializer.INSTANCE); + } + + private static List<MetadataSerializer> 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, + @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 00000000000..cf6dc8120fa --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBaseTest.java @@ -0,0 +1,443 @@ +/* + * 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}. + * + * <p>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(<exclusiveDir>, 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. + * + * <p>Throughout this test an <i>own</i> handle is one whose file lives in the directory the + * metadata is written to; a <i>foreign</i> handle points anywhere else. + */ +class MetadataV2V3SerializerBaseTest { + + @TempDir private java.nio.file.Path tmp; + + /** + * A shared-state file that lives <b>outside</b> 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 relative encoding must be kept + * unconditionally: every relative handle stays relative and is resolved against whatever + * directory the metadata is later read from. + */ + @Test + void testNullExclusiveDirKeepsRelativeEncoding() 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)) { + // no exclusive directory: every relative handle stays relative, unconditionally + MetadataV3Serializer.INSTANCE.serialize( + metadataFor(incrementalHandleWithSharedState(foreignHandle, "000001.sst")), + dos, + null); + } + + 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("relative 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, + MetadataV2V3SerializerBase.SerializationContext.withExclusiveDir( + 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 no-exclusive-directory 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 withExclusiveDir = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(withExclusiveDir)) { + MetadataV2V3SerializerBase.serializeStreamStateHandle( + ownHandle, + dos, + MetadataV2V3SerializerBase.SerializationContext.withExclusiveDir( + exclusiveDir)); + } + final ByteArrayOutputStream withoutExclusiveDir = new ByteArrayOutputStream(); + try (DataOutputStream dos = new DataOutputStream(withoutExclusiveDir)) { + MetadataV2V3SerializerBase.serializeStreamStateHandle( + ownHandle, + dos, + MetadataV2V3SerializerBase.SerializationContext.withoutExclusiveDir()); + } + + assertThat(withExclusiveDir.toByteArray()) + .as(scheme + " own handle must keep the relative encoding") + .isEqualTo(withoutExclusiveDir.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<HandleAndLocalPath> 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 e3db858ed9e..543f15b500b 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 @@ -219,7 +219,7 @@ class MetadataV3SerializerTest { CheckpointMetadata metadata = new CheckpointMetadata(checkpointId, operatorStates, masterStates); - MetadataV3Serializer.INSTANCE.serialize(metadata, out); + MetadataV3Serializer.INSTANCE.serialize(metadata, out, null); out.close(); // The relative pointer resolution in MetadataV2V3SerializerBase currently runs the same @@ -268,7 +268,8 @@ class MetadataV3SerializerTest { try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) { MetadataV2V3SerializerBase.serializeStreamStateHandle( new KeyGroupsStateHandle(offsets, new ByteStreamStateHandle("test", data)), - new DataOutputStream(out)); + new DataOutputStream(out), + MetadataV2V3SerializerBase.SerializationContext.withoutExclusiveDir()); try (ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray())) { StreamStateHandle handle = MetadataV2V3SerializerBase.deserializeStreamStateHandle( @@ -298,7 +299,10 @@ class MetadataV3SerializerTest { 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), + MetadataV2V3SerializerBase.SerializationContext.withoutExclusiveDir()); try (ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray())) { KeyedStateHandle deserialized = MetadataV2V3SerializerBase.deserializeKeyedStateHandle( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4SerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4SerializerTest.java index 79c4329c236..0d7bcd53559 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4SerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV4SerializerTest.java @@ -47,7 +47,7 @@ class MetadataV4SerializerTest { try (ByteArrayOutputStream out = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(out)) { - instance.serialize(metadata, dos); + instance.serialize(metadata, dos, null); try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(out.toByteArray()))) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5SerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5SerializerTest.java index f0a82a01124..43fe649f4fa 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5SerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV5SerializerTest.java @@ -59,7 +59,7 @@ class MetadataV5SerializerTest { void testSerializeOperatorUidAndName() throws IOException { try (ByteArrayOutputStream out = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(out)) { - INSTANCE.serialize(metadata, dos); + INSTANCE.serialize(metadata, dos, null); try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(out.toByteArray()))) { @@ -94,7 +94,7 @@ class MetadataV5SerializerTest { final IllegalArgumentException exception = assertThrows( IllegalArgumentException.class, - () -> INSTANCE.serialize(metadata, dos)); + () -> INSTANCE.serialize(metadata, dos, null)); assertThat(exception.getMessage()).contains(exceptionMessage); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV6SerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV6SerializerTest.java index 605a47c2794..cea61cf6443 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV6SerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV6SerializerTest.java @@ -103,7 +103,7 @@ class MetadataV6SerializerTest { try (ByteArrayOutputStream out = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(out)) { - INSTANCE.serialize(metadata, dos); + INSTANCE.serialize(metadata, dos, null); try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(out.toByteArray()))) { 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 0b51a13734c..49ca4c54ea1 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 class TestUtils { 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 0bcf91e0654..ffa8e212756 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 @@ class OperatorCoordinatorSchedulerTest { 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 00000000000..02a2f7beccb --- /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. + * + * <p>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 <i>savepoint</i> + * directory, not in the checkpoint's own exclusive directory. + * + * <p>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/<uuid>} while + * the bytes live at {@code savepointDir/<uuid>}, and restoring from the checkpoint would fail with + * a file-not-found error. + * + * <p>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<String> 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<Integer> 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<Integer> 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/<uuid> + // (which does not exist) instead of savepointDir/<uuid>. + List<File> 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<Integer> 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<Integer> buildBackend( + File instanceBasePath, Collection<KeyedStateHandle> 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<Integer> backend, + ValueStateDescriptor<String> descriptor, + int fromInclusive, + int toExclusive) + throws Exception { + for (int key = fromInclusive; key < toExclusive; key++) { + backend.setCurrentKey(key); + ValueState<String> state = + backend.getPartitionedState( + VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, descriptor); + state.update(valueFor(key)); + } + } + + private void assertStateRestored( + RocksDBKeyedStateBackend<Integer> backend, + ValueStateDescriptor<String> descriptor, + int fromInclusive, + int toExclusive) + throws Exception { + for (int key = fromInclusive; key < toExclusive; key++) { + backend.setCurrentKey(key); + ValueState<String> 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<Integer> backend, + long checkpointId, + FsCheckpointStreamFactory streamFactory, + CheckpointOptions options) + throws Exception { + RunnableFuture<SnapshotResult<KeyedStateHandle>> 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 00000000000..a6f83fa94c7 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointAfterClaimedNativeSavepointITCase.java @@ -0,0 +1,280 @@ +/* + * 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.state.rocksdb.RocksDBOptions; +import org.apache.flink.state.rocksdb.RocksDBOptionsFactory; +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 org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; + +import java.nio.file.Path; +import java.util.Collection; +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.runtime.testutils.CommonTestUtils.waitUntilCondition; +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. + * + * <p>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. + * + * <p>{@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); + // Disable RocksDB auto-compaction so the SSTs inherited from the claimed savepoint are not + // compacted away before the first post-restore checkpoint references them. + config.set( + RocksDBOptions.OPTIONS_FACTORY, + DisableAutoCompactionOptionsFactory.class.getName()); + 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(); + + // Take checkpoints until the running job actually has SST files in its shared state, so + // the NATIVE savepoint below inherits real SSTs. A checkpoint racing ahead of the first + // processed records would flush an empty memtable and produce none, so retry the + // checkpoint until SSTs appear. + final JobGraph initialJob = createJobGraph(config); + client.submitJob(initialJob).get(); + waitForAllTaskRunning(miniCluster, initialJob.getJobID(), false); + waitUntilCondition( + () -> + !sharedStateFilePaths( + miniCluster + .triggerCheckpoint(initialJob.getJobID()) + .get()) + .isEmpty()); + 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<String> 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(); + } + + /** + * Paces the source so checkpoint barriers do not queue behind a large record backlog. Keeping + * the reused savepoint SSTs alive is handled deterministically by disabling auto-compaction + * (see {@link DisableAutoCompactionOptionsFactory}), not by this throttle. + */ + private static final class Throttler implements MapFunction<Long, Long> { + + 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<Long, Long> { + + private ValueState<Long> 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; + } + } + + /** + * Disables RocksDB auto-compaction so the SSTs inherited from the claimed savepoint survive + * until the first post-restore checkpoint references them. Referenced by class name through + * {@link RocksDBOptions#OPTIONS_FACTORY}, so it must stay public with a no-arg constructor. + */ + public static final class DisableAutoCompactionOptionsFactory implements RocksDBOptionsFactory { + + private static final long serialVersionUID = 1L; + + @Override + public DBOptions createDBOptions( + DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) { + return currentOptions; + } + + @Override + public ColumnFamilyOptions createColumnOptions( + ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) { + return currentOptions.setDisableAutoCompactions(true); + } + } +}
