[FLINK-39964][checkpoint] Fix unrestorable checkpoint after CLAIM restore from native savepoint

flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java

@@ -79,7 +79,19 @@ public void writeRecord(CheckpointMetadata metadata) throws IOException {
                         () -> {
                             try (CheckpointMetadataOutputStream out =
                                     targetLocation.createMetadataOutputStream()) {
-                                Checkpoints.storeCheckpointMetadata(metadata, out);
+                                // Must stay on the variant WITHOUT the exclusive directory. Files
+                                // retained from an existing savepoint are copied into this
+                                // directory by FileCopyFunction, but their handles still reference
+                                // the source savepoint. Writing without the exclusive directory
+                                // keeps the relative (file-name-only) encoding, which resolves
+                                // against this directory on restore and keeps the savepoint
+                                // self-contained. The exclusive-dir-aware
+                                // Checkpoints.storeCheckpointMetadata would instead fill this
+                                // savepoint's metadata with absolute paths pointing into the
+                                // source savepoint, breaking this savepoint once the source is
+                                // deleted.
+                                Checkpoints.storeCheckpointMetadataWithoutExclusiveDir(
+                                        metadata, out);
                                 CompletedCheckpointStorageLocation finalizedLocation =
                                         out.closeAndFinalizeCheckpoint();
                                 return finalizedLocation.getExternalPointer();

flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java

@@ -35,6 +35,7 @@
 import org.apache.flink.test.util.AbstractTestBaseJUnit4;
 import org.apache.flink.util.AbstractID;
 import org.apache.flink.util.Collector;
+import org.apache.flink.util.FileUtils;
 
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.Assert;
@@ -173,7 +174,7 @@ public void testSavepointDeepCopy() throws Exception {
                 .write(savepointPath2);
         env.execute("create savepoint2");
 
-        Set<String> stateFiles2 = getFileNamesInDirectory(Paths.get(savepointPath1));
+        Set<String> stateFiles2 = getFileNamesInDirectory(Paths.get(savepointPath2));
 
         Assert.assertTrue(
                 "Failed to create savepoint2 from savepoint1 with additional state files",
@@ -184,6 +185,11 @@ public void testSavepointDeepCopy() throws Exception {
                 stateFiles1,
                 everyItem(isIn(stateFiles2)));
 
+        // Not a cleanup step: deleting savepoint1 before reading savepoint2 proves the deep
+        // copy is self-contained — savepoint2's metadata must reference the copied files, not
+        // the originals in savepoint1.
+        FileUtils.deleteDirectory(new File(savepointPath1));
+
         // Try to fromExistingSavepoint savepoint2 and read the state of "Operator1" (which has not
         // been
         // touched/changed when savepoint2

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}. This test is the
+ * fast regression guard for that choice; {@code SavepointDeepCopyTest} covers the same guarantee
+ * end-to-end.
+ */
+class SavepointOutputFormatSelfContainedTest {
+
+    private static final int PARALLELISM = 1;
+    private static final int MAX_PARALLELISM = 128;
+    private static final int SUBTASK_INDEX = 0;
+    private static final long CHECKPOINT_ID = 1L;
+    private static final String OPERATOR_UID = "uid";
+    private static final String STATE_FILE_NAME = "sst-0001.state";
+
+    @Test
+    void testSavepointWithRetainedStateStaysSelfContained(@TempDir File tempDir) throws Exception {
+        // (1) a source savepoint directory ('sp-old') and the new one being written ('sp-new').
+        File sourceSavepointDir = new File(tempDir, "sp-old");
+        File targetSavepointDir = new File(tempDir, "sp-new");
+        assertThat(sourceSavepointDir.mkdirs()).isTrue();
+        assertThat(targetSavepointDir.mkdirs()).isTrue();
+
+        // (2) a retained state file living in the source savepoint, referenced by a relative
+        // handle whose full path still points at the file in the source directory (this is what
+        // loading an existing savepoint produces).
+        byte[] stateBytes = "retained-shared-state".getBytes(StandardCharsets.UTF_8);
+        File sourceStateFile = new File(sourceSavepointDir, STATE_FILE_NAME);
+        Files.write(sourceStateFile.toPath(), stateBytes);
+
+        Path targetSavepointPath = new Path(targetSavepointDir.getAbsolutePath());
+        Path sourceStatePath =
+                new Path(new Path(sourceSavepointDir.getAbsolutePath()), STATE_FILE_NAME);
+        RelativeFileStateHandle retainedHandle =
+                new RelativeFileStateHandle(sourceStatePath, STATE_FILE_NAME, stateBytes.length);
+
+        // (3) simulate FileCopyFunction copying the file into the new savepoint directory under
+        // the same name.
+        Files.copy(
+                sourceStateFile.toPath(), new File(targetSavepointDir, STATE_FILE_NAME).toPath());
+
+        // (4) build metadata that carries the retained relative handle as shared incremental state.
+        CheckpointMetadata metadata = createMetadata(retainedHandle);
+
+        // (5) write the savepoint metadata into the new directory.
+        SavepointOutputFormat format = new SavepointOutputFormat(targetSavepointPath);
+        format.setRuntimeContext(new MockStreamingRuntimeContext(PARALLELISM, SUBTASK_INDEX));
+        format.open(FirstAttemptInitializationContext.of(SUBTASK_INDEX, PARALLELISM));
+        format.writeRecord(metadata);
+        format.close();
+
+        // (6) reload the written metadata, resolving relative handles against the new directory.
+        CheckpointMetadata reloaded =
+                SavepointLoader.loadSavepointMetadata(targetSavepointPath.getPath());
+        StreamStateHandle reloadedShared = extractSingleSharedStateHandle(reloaded);
+
+        // (7) the reloaded reference must point at the copy inside the new savepoint, not back at
+        // the original file in the source savepoint.
+        assertThat(reloadedShared)
+                .as(
+                        "Retained state must reload as a relative handle that resolves inside "
+                                + "the new savepoint; an absolute path back into the source "
+                                + "savepoint would break the new savepoint once the source is "
+                                + "deleted")
+                .isInstanceOf(RelativeFileStateHandle.class);
+
+        RelativeFileStateHandle reloadedRelative = (RelativeFileStateHandle) reloadedShared;
+        assertThat(reloadedRelative.getRelativePath()).isEqualTo(STATE_FILE_NAME);
+        assertThat(reloadedRelative.getFilePath().getPath())
+                .isEqualTo(new Path(targetSavepointPath, STATE_FILE_NAME).getPath());
+        assertThat(new File(reloadedRelative.getFilePath().getPath())).exists();
+    }
+
+    private static CheckpointMetadata createMetadata(RelativeFileStateHandle retainedHandle) {
+        StreamStateHandle metaStateHandle =
+                new ByteStreamStateHandle(
+                        "backend-meta", "backend-meta".getBytes(StandardCharsets.UTF_8));
+        IncrementalRemoteKeyedStateHandle keyedStateHandle =
+                new IncrementalRemoteKeyedStateHandle(
+                        UUID.randomUUID(),
+                        KeyGroupRange.of(0, MAX_PARALLELISM - 1),
+                        CHECKPOINT_ID,
+                        Collections.singletonList(
+                                HandleAndLocalPath.of(retainedHandle, STATE_FILE_NAME)),
+                        Collections.emptyList(),
+                        metaStateHandle);
+
+        OperatorSubtaskState subtaskState =
+                OperatorSubtaskState.builder().setManagedKeyedState(keyedStateHandle).build();
+
+        OperatorState operatorState =
+                new OperatorState(
+                        null,
+                        OPERATOR_UID,
+                        OperatorIDGenerator.fromUid(OPERATOR_UID),
+                        PARALLELISM,
+                        MAX_PARALLELISM);
+        operatorState.putState(SUBTASK_INDEX, subtaskState);
+
+        return new CheckpointMetadata(
+                CHECKPOINT_ID, Collections.singleton(operatorState), Collections.emptyList());
+    }
+
+    private static StreamStateHandle extractSingleSharedStateHandle(CheckpointMetadata metadata) {
+        IncrementalRemoteKeyedStateHandle keyedStateHandle =
+                (IncrementalRemoteKeyedStateHandle)
+                        metadata.getOperatorStates()
+                                .iterator()
+                                .next()
+                                .getState(SUBTASK_INDEX)
+                                .getManagedKeyedState()
+                                .iterator()
+                                .next();
+        return keyedStateHandle.getSharedState().get(0).getHandle();
+    }
+}

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.OperatorIDPair;
 import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
 import org.apache.flink.runtime.checkpoint.metadata.MetadataSerializer;
@@ -28,6 +29,7 @@
 import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointMetadataOutputStream;
 import org.apache.flink.runtime.state.CheckpointStorage;
 import org.apache.flink.runtime.state.CheckpointStorageLoader;
 import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
@@ -76,29 +78,80 @@ public class Checkpoints {
     //  Writing out checkpoint metadata
     // ------------------------------------------------------------------------
 
-    public static void storeCheckpointMetadata(
+    /**
+     * Stores the checkpoint metadata without knowing the checkpoint's exclusive directory: relative
+     * file references are always persisted as relative and are resolved against whatever directory
+     * the metadata is later read from. When writing the metadata of an actual checkpoint or
+     * savepoint, prefer {@link #storeCheckpointMetadata(CheckpointMetadata,
+     * CheckpointMetadataOutputStream)}.
+     *
+     * <p>The deliberately different method name keeps the context-free variants out of the {@code
+     * storeCheckpointMetadata} overload set, so a caller cannot switch between the two encodings by
+     * merely changing a variable's static type.
+     */
+    public static void storeCheckpointMetadataWithoutExclusiveDir(
             CheckpointMetadata checkpointMetadata, OutputStream out) throws IOException {
 
         DataOutputStream dos = new DataOutputStream(out);
-        storeCheckpointMetadata(checkpointMetadata, dos);
+        storeCheckpointMetadataWithoutExclusiveDir(checkpointMetadata, dos);
     }
 
+    /**
+     * Stores the checkpoint metadata into the given metadata output stream, passing the stream's
+     * exclusive checkpoint directory to the serializer so that relative file references stay
+     * self-consistent on recovery.
+     */
     public static void storeCheckpointMetadata(
+            CheckpointMetadata checkpointMetadata, CheckpointMetadataOutputStream out)
+            throws IOException {
+
+        DataOutputStream dos = new DataOutputStream(out);
+        storeCheckpointMetadata(
+                checkpointMetadata,
+                dos,
+                MetadataV6Serializer.INSTANCE,
+                out.getExclusiveCheckpointDir());
+    }
+
+    /**
+     * Variant of {@link #storeCheckpointMetadataWithoutExclusiveDir(CheckpointMetadata,
+     * OutputStream)} for a {@link DataOutputStream}.
+     */
+    public static void storeCheckpointMetadataWithoutExclusiveDir(
             CheckpointMetadata checkpointMetadata, DataOutputStream out) throws IOException {
-        storeCheckpointMetadata(checkpointMetadata, out, MetadataV6Serializer.INSTANCE);
+        storeCheckpointMetadataWithoutExclusiveDir(
+                checkpointMetadata, out, MetadataV6Serializer.INSTANCE);
     }
 
     public static void storeCheckpointMetadata(
+            CheckpointMetadata checkpointMetadata,
+            DataOutputStream out,
+            @Nullable Path exclusiveDir)
+            throws IOException {
+        storeCheckpointMetadata(
+                checkpointMetadata, out, MetadataV6Serializer.INSTANCE, exclusiveDir);
+    }
+
+    public static void storeCheckpointMetadataWithoutExclusiveDir(
             CheckpointMetadata checkpointMetadata,
             DataOutputStream out,
             MetadataSerializer serializer)
             throws IOException {
+        storeCheckpointMetadata(checkpointMetadata, out, serializer, null);
+    }
+
+    public static void storeCheckpointMetadata(
+            CheckpointMetadata checkpointMetadata,
+            DataOutputStream out,
+            MetadataSerializer serializer,
+            @Nullable Path exclusiveDir)
+            throws IOException {
 
         // write generic header
         out.writeInt(HEADER_MAGIC_NUMBER);
 
         out.writeInt(serializer.getVersion());
-        serializer.serialize(checkpointMetadata, out);
+        serializer.serialize(checkpointMetadata, out, exclusiveDir);
     }
 
     // ------------------------------------------------------------------------

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java

@@ -18,8 +18,11 @@
 
 package org.apache.flink.runtime.checkpoint.metadata;
 
+import org.apache.flink.core.fs.Path;
 import org.apache.flink.core.io.Versioned;
 
+import javax.annotation.Nullable;
+
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -46,9 +49,42 @@ CheckpointMetadata deserialize(
             throws IOException;
 
     /**
-     * Serializes a savepoint or checkpoint metadata to an output stream.
+     * Serializes a savepoint or checkpoint metadata to an output stream without knowledge of the
+     * exclusive directory the metadata is written into.
+     *
+     * <p>This is a convenience variant of {@link #serialize(CheckpointMetadata, DataOutputStream,
+     * Path)} with a {@code null} exclusive directory: relative file references keep the relative
+     * encoding unconditionally and are resolved against whatever directory the metadata is later
+     * read from, which is only correct if every referenced file actually lives in that directory.
+     *
+     * @throws IOException Serialization failures are forwarded
+     */
+    default void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos)
+            throws IOException {
+        serialize(checkpointMetadata, dos, null);
+    }
+
+    /**
+     * Serializes a savepoint or checkpoint metadata to an output stream, knowing the exclusive
+     * directory into which the metadata (and the checkpoint/savepoint it describes) is being
+     * written.
+     *
+     * <p>The exclusive directory is used to keep the on-disk references self-consistent: a relative
+     * file handle that does not belong to {@code exclusiveDirPath} (e.g. a savepoint SST reused by
+     * the first incremental checkpoint after a CLAIM-mode restore) is persisted with its absolute
+     * path instead of being (incorrectly) resolved against {@code exclusiveDirPath} on recovery.
      *
+     * <p>This is the method implementations must provide (rather than a default dropping the
+     * exclusive directory), so that a serializer cannot silently lose the exclusive-directory
+     * information.
+     *
+     * @param exclusiveDirPath the exclusive directory of the checkpoint/savepoint being written, or
+     *     {@code null} if it is unknown.
      * @throws IOException Serialization failures are forwarded
      */
-    void serialize(CheckpointMetadata checkpointMetadata, DataOutputStream dos) throws IOException;
+    void serialize(
+            CheckpointMetadata checkpointMetadata,
+            DataOutputStream dos,
+            @Nullable Path exclusiveDirPath)
+            throws IOException;
 }