pnowojski commented on code in PR #28709:
URL: https://github.com/apache/flink/pull/28709#discussion_r3585380373
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointDeepCopyTest.java:
##########
@@ -173,7 +173,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));
Review Comment:
Does this expose the bug you are fixing in the latter commit?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java:
##########
@@ -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.
+ */
Review Comment:
After reading those two java docs, I don't understand the difference between
the two methods.
What is the difference between:
- relative file references are always persisted as relative and are resolved
against whatever directory the metadata is later read from
- relative file references stay self-consistent on recovery.
?
Maybe me and this java doc is lacking explanation of what are exclusive
directories?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java:
##########
@@ -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(
Review Comment:
why do we need to keep the `without excelusive dir` variant? 🤔 What is/are
the use cases justifying it?
I would expect that every checkpoint we are writing has it's own exclusive
directory, and incorrectly referencing relative files outside of that should be
explicitly forbidden?
##########
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);
Review Comment:
I don't understand this change and this comment doesn't help me 😓
1.
> Files retained from an existing savepoint are copied into this directory
by FileCopyFunction
`FileCopyFunction` what's that? Who is copying what? `SavepointOutputFormat`
doesn't have to be invovled in copy any files AFAIU.
2.
Looking at the name, `Checkpoints.storeCheckpointMetadata ` vs
`Checkpoints.storeCheckpointMetadataWithoutExclusiveDir`, I don't understand
how that connects to:
> storeCheckpointMetadata would instead fill this savepoint's metadata with
absolute paths pointing into the source savepoint
> breaking this savepoint once the source is deleted.
Again I'm confused who is copying what, and why are we talking about
`source` in this class/method.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java:
##########
@@ -325,6 +340,14 @@ protected OperatorSubtaskState deserializeSubtaskState(
@VisibleForTesting
static void serializeKeyedStateHandle(KeyedStateHandle stateHandle,
DataOutputStream dos)
throws IOException {
+ serializeKeyedStateHandle(stateHandle, dos, null);
+ }
Review Comment:
optional: wouldn't it be safer to just drop this variant and force everyone
to expcilitly pass `null` or some concrete value?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataSerializer.java:
##########
@@ -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.
Review Comment:
When exclusive dir is `null`, and file resides outside of the `_metadata`'s
directory, shouldn't we support relative paths that are pointing away? For
example files structure:
```
/bar/claimed-files/chk-1/_old_metadata
/bar/claimed-files/chk-1/foo/1.sst
/bar/checkpoints/chk-2/_metadata
```
AFAIU the problem in this bug is that `1.sst` would be referenced in
`_metadata` as `foo/1.sst` which would be incorrect. And with your fix, we
would store `/bar/claimed-files/chk-1/foo/1.sst`. Shouldn't we instead use
`../../claimed-files/chk-1/foo/1.sst`? 🤔
But maybe indeed referencing claimed files by absolute path makes more
sense...
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java:
##########
@@ -282,7 +297,7 @@ public OperatorStateHandle
deserializeOperatorStateHandleUtil(DataInputStream di
@VisibleForTesting
public void serializeKeyedStateHandleUtil(KeyedStateHandle stateHandle,
DataOutputStream dos)
throws IOException {
- serializeKeyedStateHandle(stateHandle, dos);
+ serializeKeyedStateHandle(stateHandle, dos, null);
Review Comment:
ditto about `null`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java:
##########
@@ -270,7 +285,7 @@ public static StreamStateHandle
deserializeStreamStateHandle(DataInputStream dis
@VisibleForTesting
public void serializeOperatorStateHandleUtil(
OperatorStateHandle stateHandle, DataOutputStream dos) throws
IOException {
- serializeOperatorStateHandle(stateHandle, dos);
+ serializeOperatorStateHandle(stateHandle, dos, null);
Review Comment:
Why is it ok to pass `null` instead of a propper context in some of those
calls?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
