[FLINK-39964][state] 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,20 @@ public void writeRecord(CheckpointMetadata metadata) throws IOException {
                         () -> {
                             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();

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}. {@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();
+    }
+}

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

@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.OperatorIDPair;
 import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
 import org.apache.flink.runtime.checkpoint.metadata.MetadataSerializer;
@@ -28,12 +29,15 @@
 import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointMetadataOutputStream;
 import org.apache.flink.runtime.state.CheckpointStorage;
 import org.apache.flink.runtime.state.CheckpointStorageLoader;
+import org.apache.flink.runtime.state.CheckpointedStateScope;
 import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
 import org.apache.flink.runtime.state.StateBackend;
 import org.apache.flink.runtime.state.StateBackendLoader;
 import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.runtime.state.filesystem.RelativeFileStateHandle;
 import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
 import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
 import org.apache.flink.util.CollectionUtil;
@@ -64,6 +68,14 @@
  * <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);
     }
 
     // ------------------------------------------------------------------------

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

@@ -99,7 +99,9 @@ static <Info> void serializeChannelStateHandle(
             dos.writeLong(offset);
         }
         dos.writeLong(handle.getStateSize());
-        serializeStreamStateHandle(handle.getDelegate(), dos);
+        // null context: channel state is always written into the current checkpoint's own
+        // exclusive directory, see MetadataV3Serializer#serializeSubtaskState.
+        serializeStreamStateHandle(handle.getDelegate(), dos, null);
     }
 
     static <Info, Handle extends AbstractChannelStateHandle<Info>>

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

@@ -146,7 +146,9 @@ private <Info> void serializeMergedChannelStateHandle(
 
         dos.writeInt(handle.getSubtaskIndex());
         dos.writeLong(handle.getStateSize());
-        serializeStreamStateHandle(handle.getDelegate(), dos);
+        // null context: channel state is always written into the current checkpoint's own
+        // exclusive directory, see MetadataV3Serializer#serializeSubtaskState.
+        serializeStreamStateHandle(handle.getDelegate(), dos, null);
 
         dos.writeInt(handle.getSerializedChannelOffsets().length);
         dos.write(handle.getSerializedChannelOffsets());