gaborgsomogyi commented on code in PR #29422:
URL: https://github.com/apache/flink/pull/29422#discussion_r4216607655
##########
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:
Two things to mention:
- We're going to leave the old connectors behind
- This is a one time occasion to have a migration path. Any further
development in the old connectors will not be reflected in the native connector
##########
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:
I would say this support with the copy-paste is one time step and any
further enhancements in the serializer logic will break this contract. To be
specific we give users a one time path to the new connector
##########
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:
Under what circumstances can the legacy format magic collide with a valid
new format checkpoint?
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java:
##########
@@ -20,15 +20,22 @@
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link NativeS3RecoverableSerializer}. */
class NativeS3RecoverableSerializerTest {
+ private static final int LEGACY_MAGIC_NUMBER = 0x98761432;
Review Comment:
This already exists
--
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]
