gaborgsomogyi commented on code in PR #29422:
URL: https://github.com/apache/flink/pull/29422#discussion_r4284717704
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -108,6 +134,12 @@ public NativeS3Recoverable deserialize(int version, byte[]
serialized) throws IO
if (version != CURRENT_VERSION) {
throw new IOException("Unsupported version: " + version);
}
+ if (isLegacyFormat(serialized)) {
+ LOG.info(
Review Comment:
This INFO fires once per decoded recoverable, so a large restore logs it
many times per TaskManager. Could it go to DEBUG?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +167,119 @@ public NativeS3Recoverable deserialize(int version,
byte[] serialized) throws IO
incompleteObjectName,
incompleteObjectLength);
}
+
+ /**
+ * Checks whether the given serialized state was written by the {@code
S3RecoverableSerializer}
+ * of flink-s3-fs-base (used by {@code flink-s3-fs-hadoop} / {@code
flink-s3-fs-presto}).
+ *
+ * <p>The detection is unambiguous: the current format starts with a
{@code writeUTF} length
+ * (two bytes, big-endian) for the object name, so mistaking it for the
legacy magic number
+ * would require an object name of 0x3214 (12820) bytes, far beyond the
1024-byte S3 key limit.
+ * It is also structurally impossible: read as the little-endian magic
number, the current
+ * format's first four bytes would have to end in 0x98, which is never a
valid UTF-8 character
+ * start byte.
+ */
+ private static boolean isLegacyFormat(byte[] serialized) {
+ if (serialized.length < Integer.BYTES) {
+ return false;
+ }
+ return
ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN).getInt()
+ == LEGACY_MAGIC_NUMBER;
+ }
+
+ /**
+ * Decodes the legacy little-endian format written by {@code
S3RecoverableSerializer} in
+ * flink-s3-fs-base: magic number, then length-prefixed UTF-8 object name
and upload id, part
+ * count with {partNumber, length-prefixed eTag} pairs, numBytesInParts,
length-prefixed
+ * incomplete object name (0 length meaning none), and incomplete object
length.
+ */
+ private static NativeS3Recoverable deserializeLegacy(byte[] serialized)
throws IOException {
+ try {
+ return decodeLegacy(serialized);
+ } catch (BufferUnderflowException
+ | NegativeArraySizeException
+ | IllegalArgumentException e) {
+ throw new IOException(
+ "Corrupt or truncated S3 multipart upload state in legacy
format "
+ + "(written by flink-s3-fs-hadoop or
flink-s3-fs-presto).",
+ e);
+ }
+ }
+
+ private static NativeS3Recoverable decodeLegacy(byte[] serialized) throws
IOException {
+ final ByteBuffer bb =
ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN);
+
+ // re-check kept so this decoder stays a faithful copy of the
authoritative
+ // S3RecoverableSerializer.deserializeV1 in flink-s3-fs-base
+ if (bb.getInt() != LEGACY_MAGIC_NUMBER) {
+ throw new IOException("Corrupt data: Unexpected magic number.");
+ }
+
+ final byte[] keyBytes = new byte[readBoundedLength(bb)];
+ bb.get(keyBytes);
+
+ final byte[] uploadIdBytes = new byte[readBoundedLength(bb)];
+ bb.get(uploadIdBytes);
+
+ final int numParts = readBoundedPartCount(bb);
Review Comment:
The decoder accepts zero parts, which is what a closed zero-byte file looks
like in legacy state. The Hadoop `S3Committer` treats that as a no-op
(`totalLength > 0`). `NativeS3Committer#commit` throws on empty parts, so such
a savepoint would fail on every restore attempt. Should empty parts be a no-op
there @Samrat002 ? We can mark this no or as follow-up if it's shaky...
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +166,116 @@ public NativeS3Recoverable deserialize(int version,
byte[] serialized) throws IO
incompleteObjectName,
incompleteObjectLength);
}
+
+ /**
+ * Checks whether the given serialized state was written by the {@code
S3RecoverableSerializer}
+ * of flink-s3-fs-base (used by {@code flink-s3-fs-hadoop} / {@code
flink-s3-fs-presto}).
+ *
+ * <p>The detection is unambiguous: the current format starts with a
{@code writeUTF} length
+ * (two bytes, big-endian) for the object name, so mistaking it for the
legacy magic number
+ * would require an object name of 0x3214 (12820) bytes, far beyond the
1024-byte S3 key limit.
+ */
+ private static boolean isLegacyFormat(byte[] serialized) {
+ if (serialized.length < Integer.BYTES) {
+ return false;
+ }
+ return
ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN).getInt()
+ == LEGACY_MAGIC_NUMBER;
+ }
Review Comment:
Double checked, this is fine
##########
docs/content/docs/deployment/filesystems/s3.md:
##########
@@ -153,6 +153,10 @@ The legacy configuration key `s3.path.style.access` is
still supported as a fall
The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2
completely removing the dependency on Hadoop. It is registered under the
schemes *s3://* and *s3a://*. It provides a drop-in replacement for the Presto
and Hadoop implementations, supporting checkpointing, the [FileSink]({{< ref
"docs/connectors/datastream/filesystem" >}}) (via `RecoverableWriter`),
server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role
assumption, entropy injection, and bulk copy via S3TransferManager.
+{{< hint info >}}
+**Switching from the Hadoop or Presto implementations:** jobs with pending or
in-progress `FileSink` part files in their state can be restored directly with
the Native S3 FileSystem — checkpoints and savepoints written by the
Hadoop/Presto recoverable-state format are detected and migrated on restore.
The reverse direction is **not** supported: state written by the Native S3
FileSystem cannot be restored with the Hadoop or Presto implementations. Drain
the job first (e.g. `stop-with-savepoint --drain`) before switching back.
+{{< /hint >}}
Review Comment:
Thanks, both points are in. Could we split this into short sentences and
name only Hadoop (Presto can't write `FileSink` state)? Suggested text below.
The drain sentence is left out on purpose, see my question in the conversation.
```
{{< hint info >}}
**Switching from the Hadoop implementation:** jobs with pending or
in-progress `FileSink` part files in their state can be restored with the
Native S3 FileSystem, which detects and decodes the state written by the Hadoop
implementation.
This is a one-time migration path. The Hadoop implementation is being left
behind and later changes to its state format are not followed. The reverse
direction is not supported: state written by the Native S3 FileSystem cannot be
restored with the Hadoop implementation.
{{< /hint >}}
```
##########
flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java:
##########
@@ -41,6 +41,10 @@ final class S3RecoverableSerializer implements
SimpleVersionedSerializer<S3Recov
private static final int MAGIC_NUMBER = 0x98761432;
+ // The wire format below is frozen: flink-s3-fs-native
(NativeS3RecoverableSerializer)
+ // duplicates this decoder to restore state after switching to that plugin
(FLINK-40943).
+ // Do not change the format without keeping that decoder in sync.
Review Comment:
Correct, we're going to leave the old connectors behind and I don't want a
version matrix to maintain: the native decoder is a one-time copy, and if the
Hadoop format ever changes, that's it. Could you therefore drop `Do not change
the format without keeping that decoder in sync` from the comment, e.g.
"One-time copy: flink-s3-fs-native duplicates this decoder (FLINK-40943), later
format changes are not mirrored there"?
##########
docs/content/docs/deployment/filesystems/s3.md:
##########
@@ -153,6 +153,10 @@ The legacy configuration key `s3.path.style.access` is
still supported as a fall
The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2
completely removing the dependency on Hadoop. It is registered under the
schemes *s3://* and *s3a://*. It provides a drop-in replacement for the Presto
and Hadoop implementations, supporting checkpointing, the [FileSink]({{< ref
"docs/connectors/datastream/filesystem" >}}) (via `RecoverableWriter`),
server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role
assumption, entropy injection, and bulk copy via S3TransferManager.
+{{< hint info >}}
+**Switching from the Hadoop or Presto implementations:** jobs with pending or
in-progress `FileSink` part files in their state can be restored directly with
the Native S3 FileSystem — checkpoints and savepoints written by the
Hadoop/Presto recoverable-state format are detected and migrated on restore.
The Hadoop and Presto implementations are being left behind, and this migration
path is a one-time occasion as it covers the legacy state format as it exists
today, and any future change to the Hadoop/Presto state format could not be
picked up by the Native S3 FileSystem. The reverse direction is also **not**
supported: state written by the Native S3 FileSystem cannot be restored with
the Hadoop or Presto implementations. Drain the job first (e.g.
`stop-with-savepoint --drain`) before switching back.
Review Comment:
Seems like I just wanted to ask that `.../content.zh/...` file needs the
same changes.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -108,6 +134,12 @@ public NativeS3Recoverable deserialize(int version, byte[]
serialized) throws IO
if (version != CURRENT_VERSION) {
throw new IOException("Unsupported version: " + version);
}
+ if (isLegacyFormat(serialized)) {
+ LOG.info(
+ "Detected S3 multipart upload state written by
flink-s3-fs-hadoop or "
+ + "flink-s3-fs-presto (legacy format); restoring
it with the native S3 filesystem.");
Review Comment:
`flink-s3-fs-presto` has no `RecoverableWriter`
(`FlinkS3FileSystem#createRecoverableWriter` throws for it), so it can't have
written this state. Could we drop Presto from this message and from the other
places that mention it (the Javadoc, the exception message, the docs hint and
the PR description)?
--
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]
