This is an automated email from the ASF dual-hosted git repository.
Samrat002 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 2ea9ccbc3e3 [FLINK-40943][filesystems] Restore FileSink state written
with flink-s3-fs-hadoop in flink-s3-fs-native
2ea9ccbc3e3 is described below
commit 2ea9ccbc3e3436f320a1345794c0e3a3028f7f3b
Author: Joao Boto <[email protected]>
AuthorDate: Tue Nov 3 16:32:01 2026 +0100
[FLINK-40943][filesystems] Restore FileSink state written with
flink-s3-fs-hadoop in flink-s3-fs-native
---
docs/content.zh/docs/deployment/filesystems/s3.md | 8 +
docs/content/docs/deployment/filesystems/s3.md | 8 +
.../s3/common/writer/S3RecoverableSerializer.java | 3 +
.../writer/NativeS3RecoverableSerializer.java | 127 +++++++
.../writer/NativeS3RecoverableSerializerTest.java | 367 +++++++++++++++++++++
5 files changed, 513 insertions(+)
diff --git a/docs/content.zh/docs/deployment/filesystems/s3.md
b/docs/content.zh/docs/deployment/filesystems/s3.md
index 27d4a77bde2..642d25226e6 100644
--- a/docs/content.zh/docs/deployment/filesystems/s3.md
+++ b/docs/content.zh/docs/deployment/filesystems/s3.md
@@ -153,6 +153,14 @@ 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 b [...]
+{{< 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.
+
+Known gap: a part file that was closed with zero bytes in the Hadoop format
has no committed parts. The Native S3 FileSystem cannot commit it, so restoring
such state fails.
+{{< /hint >}}
+
#### Cleaning Up Unfinished Uploads
The Native S3 FileSystem never aborts a multipart upload on its own. An upload
that is in progress may be referenced by a checkpoint or savepoint, and Flink
cannot tell whether that is the case when a stream is closed, cancelled, or
fails to commit. Aborting such an upload would make the referencing checkpoint
or savepoint unrecoverable. Uploads that are no longer needed (for example
after a job is cancelled and never restored, or after a failed commit)
therefore remain in the bucket and [...]
diff --git a/docs/content/docs/deployment/filesystems/s3.md
b/docs/content/docs/deployment/filesystems/s3.md
index 40756c5d95c..07903d08b56 100644
--- a/docs/content/docs/deployment/filesystems/s3.md
+++ b/docs/content/docs/deployment/filesystems/s3.md
@@ -153,6 +153,14 @@ 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 b [...]
+{{< 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.
+
+Known gap: a part file that was closed with zero bytes in the Hadoop format
has no committed parts. The Native S3 FileSystem cannot commit it, so restoring
such state fails.
+{{< /hint >}}
+
#### Cleaning Up Unfinished Uploads
The Native S3 FileSystem never aborts a multipart upload on its own. An upload
that is in progress may be referenced by a checkpoint or savepoint, and Flink
cannot tell whether that is the case when a stream is closed, cancelled, or
fails to commit. Aborting such an upload would make the referencing checkpoint
or savepoint unrecoverable. Uploads that are no longer needed (for example
after a job is cancelled and never restored, or after a failed commit)
therefore remain in the bucket and [...]
diff --git
a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
index a72e99be831..d477057bdd4 100644
---
a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
+++
b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
@@ -41,6 +41,9 @@ final class S3RecoverableSerializer implements
SimpleVersionedSerializer<S3Recov
private static final int MAGIC_NUMBER = 0x98761432;
+ // One-time copy: flink-s3-fs-native duplicates this decoder (FLINK-40943).
+ // Later format changes are not mirrored there.
+
/** Do not instantiate, use reusable {@link #INSTANCE} instead. */
private S3RecoverableSerializer() {}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java
index 5f9c28fdb8e..bf8a1fb4546 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java
@@ -18,14 +18,23 @@
package org.apache.flink.fs.s3native.writer;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.fs.s3native.writer.NativeS3Recoverable.PartETag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
@@ -68,11 +77,21 @@ import java.util.List;
*/
class NativeS3RecoverableSerializer implements
SimpleVersionedSerializer<NativeS3Recoverable> {
+ private static final Logger LOG =
LoggerFactory.getLogger(NativeS3RecoverableSerializer.class);
+
public static final NativeS3RecoverableSerializer INSTANCE =
new NativeS3RecoverableSerializer();
private static final int CURRENT_VERSION = 1;
+ private static final Charset CHARSET = StandardCharsets.UTF_8;
+
+ /**
+ * Magic number written at the beginning of the legacy ({@code
flink-s3-fs-hadoop}) serialized
+ * form, as found in {@code S3RecoverableSerializer} of flink-s3-fs-base.
+ */
+ @VisibleForTesting static final int LEGACY_MAGIC_NUMBER = 0x98761432;
+
@Override
public int getVersion() {
return CURRENT_VERSION;
@@ -108,6 +127,12 @@ class NativeS3RecoverableSerializer implements
SimpleVersionedSerializer<NativeS
if (version != CURRENT_VERSION) {
throw new IOException("Unsupported version: " + version);
}
+ if (isLegacyFormat(serialized)) {
+ LOG.debug(
+ "Detected S3 multipart upload state written by
flink-s3-fs-hadoop "
+ + "(legacy format); restoring it with the native
S3 filesystem.");
+ return deserializeLegacy(serialized);
+ }
ByteArrayInputStream inputStream = new
ByteArrayInputStream(serialized);
DataInputStream in = new DataInputStream(inputStream);
String objectName = in.readUTF();
@@ -135,4 +160,106 @@ class NativeS3RecoverableSerializer implements
SimpleVersionedSerializer<NativeS
incompleteObjectName,
incompleteObjectLength);
}
+
+ /**
+ * Returns whether the state was written by flink-s3-fs-base's {@code
S3RecoverableSerializer}.
+ */
+ 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, wrapping any decode error of corrupt state in an
{@link IOException}.
+ */
+ private static NativeS3Recoverable deserializeLegacy(byte[] serialized)
throws IOException {
+ try {
+ return decodeLegacy(serialized);
+ } catch (BufferUnderflowException
+ | NegativeArraySizeException
+ | IllegalArgumentException e) {
+ throw new IOException(
+ "Corrupt legacy S3 recoverable state: truncated or invalid
data "
+ + "(written by flink-s3-fs-hadoop).",
+ 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);
+ final List<PartETag> parts = new ArrayList<>(numParts);
+ for (int i = 0; i < numParts; i++) {
+ final int partNum = bb.getInt();
+ final byte[] buffer = new byte[readBoundedLength(bb)];
+ bb.get(buffer);
+ parts.add(new PartETag(partNum, new String(buffer, CHARSET)));
+ }
+
+ final long numBytes = bb.getLong();
+
+ final String lastPart;
+ final int lastObjectArraySize = readBoundedLength(bb);
+ if (lastObjectArraySize == 0) {
+ lastPart = null;
+ } else {
+ final byte[] lastPartBytes = new byte[lastObjectArraySize];
+ bb.get(lastPartBytes);
+ lastPart = new String(lastPartBytes, CHARSET);
+ }
+
+ final long lastPartLength = bb.getLong();
+
+ if (bb.hasRemaining()) {
+ throw new IOException(
+ "Corrupt legacy S3 recoverable state: unexpected trailing
bytes.");
+ }
+
+ return new NativeS3Recoverable(
+ new String(keyBytes, CHARSET),
+ new String(uploadIdBytes, CHARSET),
+ parts,
+ numBytes,
+ lastPart,
+ lastPartLength);
+ }
+
+ /** Reads a field length, rejecting negative values and values beyond the
remaining buffer. */
+ private static int readBoundedLength(ByteBuffer bb) throws IOException {
+ final int length = bb.getInt();
+ if (length < 0 || length > bb.remaining()) {
+ throw new IOException(
+ "Corrupt legacy S3 recoverable state: invalid field length
" + length);
+ }
+ return length;
+ }
+
+ /** Reads the part count, bounded so that a corrupt value cannot trigger a
huge allocation. */
+ private static int readBoundedPartCount(ByteBuffer bb) throws IOException {
+ final int numParts = bb.getInt();
+ // every part needs at least part number + eTag length (2 ints);
remaining also holds the
+ // trailing fields, so this bound is conservative but proportional to
the buffer size
+ if (numParts < 0 || (long) numParts * 2 * Integer.BYTES >
bb.remaining()) {
+ throw new IOException(
+ "Corrupt legacy S3 recoverable state: invalid part count "
+ numParts);
+ }
+ return numParts;
+ }
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
index c9b38954f70..74318fbcc7e 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
@@ -23,14 +23,40 @@ import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
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 {
+ /**
+ * Bytes captured once from the real {@code
S3RecoverableSerializer#serialize(S3Recoverable)} in
+ * flink-s3-fs-base (object name {@code dir/test-key-ä日🔑}, upload id
{@code upload-ä-id}, 2
+ * parts, incomplete part with a multi-byte name).
+ */
+ private static final String GOLDEN_LEGACY_HEX =
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200"
+ + "0000010000000800"
+ + "0000657461672d6f"
+ + "6e65020000000800"
+ + "0000657461672d74"
+ + "776f393000000000"
+ + "000011000000696e"
+ + "636f6d706c657465"
+ + "2d6b65792dc3a400"
+ + "04000000000000";
+
@Test
void testSerializeAndDeserializeWithParts() throws IOException {
NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
@@ -148,4 +174,345 @@ class NativeS3RecoverableSerializerTest {
assertThat(deserialized.incompleteObjectLength()).isEqualTo(1024L);
assertThat(serializer.serialize(deserialized)).isEqualTo(version1);
}
+
+ @Test
+ void testUnsupportedVersionThrows() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+ NativeS3Recoverable original =
+ new NativeS3Recoverable("key", "upload-id", new ArrayList<>(),
0L);
+ byte[] serialized = serializer.serialize(original);
+
+ assertThatThrownBy(() -> serializer.deserialize(2, serialized))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported version");
+ }
+
+ @Test
+ void testDeserializeLegacyFormatWithIncompletePart() throws IOException {
+ NativeS3Recoverable deserialized =
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+ NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ serializeLegacy(
+ "test-object-key",
+ "test-upload-id",
+ new int[] {1, 2},
+ new String[] {"etag1", "etag2"},
+ 12345L,
+ "incomplete-object-key",
+ 1024L));
+
+ assertThat(deserialized.getObjectName()).isEqualTo("test-object-key");
+ assertThat(deserialized.uploadId()).isEqualTo("test-upload-id");
+ assertThat(deserialized.numBytesInParts()).isEqualTo(12345L);
+ assertThat(deserialized.parts()).hasSize(2);
+ assertThat(deserialized.parts().get(0).getPartNumber()).isEqualTo(1);
+ assertThat(deserialized.parts().get(0).getETag()).isEqualTo("etag1");
+ assertThat(deserialized.parts().get(1).getPartNumber()).isEqualTo(2);
+ assertThat(deserialized.parts().get(1).getETag()).isEqualTo("etag2");
+
assertThat(deserialized.incompleteObjectName()).isEqualTo("incomplete-object-key");
+ assertThat(deserialized.incompleteObjectLength()).isEqualTo(1024L);
+ }
+
+ @Test
+ void testDeserializeLegacyFormatWithoutIncompletePart() throws IOException
{
+ NativeS3Recoverable deserialized =
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+ NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ serializeLegacy(
+ "test-object-key",
+ "test-upload-id",
+ new int[] {1},
+ new String[] {"etag1"},
+ 54321L,
+ null,
+ -1L));
+
+ assertThat(deserialized.getObjectName()).isEqualTo("test-object-key");
+ assertThat(deserialized.uploadId()).isEqualTo("test-upload-id");
+ assertThat(deserialized.numBytesInParts()).isEqualTo(54321L);
+ assertThat(deserialized.parts()).hasSize(1);
+ assertThat(deserialized.incompleteObjectName()).isNull();
+ assertThat(deserialized.incompleteObjectLength()).isEqualTo(-1L);
+ }
+
+ @Test
+ void testLegacyFormatWithoutParts() throws IOException {
+ NativeS3Recoverable deserialized =
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+ NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ serializeLegacy(
+ "key", "upload-id", new int[0], new String[0],
0L, null, -1L));
+
+ assertThat(deserialized.getObjectName()).isEqualTo("key");
+ assertThat(deserialized.parts()).isEmpty();
+ }
+
+ /**
+ * The current format's leading {@code writeUTF} length could numerically
collide with the
+ * legacy magic number's first two bytes for an object name of 0x3214
(12820) bytes. Even then
+ * the state must not go down the legacy path: it is decoded by the
current-format reader.
+ */
+ @Test
+ void testCurrentFormatNotMistakenForLegacyFormat() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+ String longObjectName = new String(new char[0x3214]).replace('\0',
'a');
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeUTF(longObjectName);
+ out.writeUTF("test-upload-id");
+ out.writeLong(12345L);
+ out.writeInt(0);
+ out.writeBoolean(false);
+ out.flush();
+ byte[] serialized = bytes.toByteArray();
+
+ assertThat(serialized[0]).isEqualTo((byte) 0x32);
+ assertThat(serialized[1]).isEqualTo((byte) 0x14);
+
+ NativeS3Recoverable deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+ assertThat(deserialized.getObjectName()).isEqualTo(longObjectName);
+ assertThat(deserialized.uploadId()).isEqualTo("test-upload-id");
+ assertThat(deserialized.numBytesInParts()).isEqualTo(12345L);
+ }
+
+ /**
+ * Golden test against bytes captured once from the real {@code
+ * S3RecoverableSerializer#serialize(S3Recoverable)} in flink-s3-fs-base,
pinning the legacy
+ * wire format independently of the test-side re-encoder below.
+ */
+ @Test
+ void testDeserializeGoldenLegacyBytes() throws IOException {
+ NativeS3Recoverable deserialized =
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+ NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ hexToBytes(GOLDEN_LEGACY_HEX));
+
+
assertThat(deserialized.getObjectName()).isEqualTo("dir/test-key-\u00e4\u65e5\uD83D\uDD11");
+ assertThat(deserialized.uploadId()).isEqualTo("upload-\u00e4-id");
+ assertThat(deserialized.numBytesInParts()).isEqualTo(12345L);
+ assertThat(deserialized.parts()).hasSize(2);
+ assertThat(deserialized.parts().get(0).getPartNumber()).isEqualTo(1);
+
assertThat(deserialized.parts().get(0).getETag()).isEqualTo("etag-one");
+ assertThat(deserialized.parts().get(1).getPartNumber()).isEqualTo(2);
+
assertThat(deserialized.parts().get(1).getETag()).isEqualTo("etag-two");
+
assertThat(deserialized.incompleteObjectName()).isEqualTo("incomplete-key-\u00e4");
+ assertThat(deserialized.incompleteObjectLength()).isEqualTo(1024L);
+ }
+
+ @Test
+ void testDeserializeLegacyWithMultiByteNames() throws IOException {
+ NativeS3Recoverable deserialized =
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+ NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ serializeLegacy(
+ "dir/ключ-\u00e4日\uD83D\uDD11",
+ "upload-id-\u00e9",
+ new int[] {7},
+ new String[] {"etag-ä"},
+ 99L,
+ "trailing-\u00fc",
+ 7L));
+
+
assertThat(deserialized.getObjectName()).isEqualTo("dir/ключ-\u00e4日\uD83D\uDD11");
+ assertThat(deserialized.uploadId()).isEqualTo("upload-id-\u00e9");
+ assertThat(deserialized.parts().get(0).getETag()).isEqualTo("etag-ä");
+
assertThat(deserialized.incompleteObjectName()).isEqualTo("trailing-\u00fc");
+ }
+
+ @Test
+ void testDeserializeLegacyTruncatedThrowsIOException() {
+ byte[] legacy = hexToBytes(GOLDEN_LEGACY_HEX.substring(0, 64)); //
cuts off mid upload id
+ byte[] truncated = new byte[legacy.length - 5];
+ System.arraycopy(legacy, 0, truncated, 0, truncated.length);
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ truncated))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Corrupt legacy");
+ }
+
+ @Test
+ void testDeserializeLegacyUnderflowOnFinalLengthThrowsIOException() {
+ byte[] golden = hexToBytes(GOLDEN_LEGACY_HEX);
+ byte[] truncated = new byte[golden.length - 1]; // underflows reading
the final long
+ System.arraycopy(golden, 0, truncated, 0, truncated.length);
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ truncated))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Corrupt legacy");
+ }
+
+ @Test
+ void
testDeserializeLegacyWithNonPositiveIncompleteLengthThrowsIOException()
+ throws IOException {
+ // an incomplete-part name with length <= 0 violates the recoverable's
invariants, so the
+ // constructor's checkArgument must surface as an IOException
+ byte[] legacy =
+ serializeLegacy(
+ "key", "upload-id", new int[0], new String[0], 0L,
"incomplete-key", 0L);
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ legacy))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Corrupt legacy");
+ }
+
+ @Test
+ void testDeserializeLegacyWithTrailingBytesThrowsIOException() {
+ byte[] legacy = serializeLegacyQuietly("key", "upload-id", 0L);
+ byte[] withTrailing = new byte[legacy.length + 1];
+ System.arraycopy(legacy, 0, withTrailing, 0, legacy.length);
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ withTrailing))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("trailing bytes");
+ }
+
+ @Test
+ void testDeserializeLegacyWithNegativeLengthThrowsIOException() throws
IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ putIntLE(out, NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putIntLE(out, -1);
+ byte[] legacy = out.toByteArray();
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ legacy))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid field length");
+ }
+
+ @Test
+ void testDeserializeLegacyWithHugeLengthThrowsIOException() throws
IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ putIntLE(out, NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putIntLE(out, Integer.MAX_VALUE);
+ byte[] legacy = out.toByteArray();
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ legacy))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid field length");
+ }
+
+ @Test
+ void testDeserializeLegacyWithHugePartCountThrowsIOException() throws
IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ putIntLE(out, NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putIntLE(out, 0); // object name length
+ putIntLE(out, 0); // upload id length
+ putIntLE(out, Integer.MAX_VALUE); // part count
+ byte[] legacy = out.toByteArray();
+
+ assertThatThrownBy(
+ () ->
+
NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ legacy))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid part count");
+ }
+
+ private static byte[] serializeLegacyQuietly(
+ String objectName, String uploadId, long numBytesInParts) {
+ try {
+ return serializeLegacy(
+ objectName, uploadId, new int[0], new String[0],
numBytesInParts, null, -1L);
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ private static byte[] hexToBytes(String hex) {
+ String normalized = hex.replace(" ", "");
+ byte[] bytes = new byte[normalized.length() / 2];
+ for (int i = 0; i < bytes.length; i++) {
+ int hi = Character.digit(normalized.charAt(2 * i), 16);
+ int lo = Character.digit(normalized.charAt(2 * i + 1), 16);
+ bytes[i] = (byte) ((hi << 4) + lo);
+ }
+ return bytes;
+ }
+
+ /**
+ * Reproduces the binary format written by {@code S3RecoverableSerializer}
(version 1) of
+ * flink-s3-fs-base, used by flink-s3-fs-hadoop: little-endian, leading
magic number, then
+ * length-prefixed UTF-8 strings.
+ */
+ private static byte[] serializeLegacy(
+ String objectName,
+ String uploadId,
+ int[] partNumbers,
+ String[] eTags,
+ long numBytesInParts,
+ String incompleteObjectName,
+ long incompleteObjectLength)
+ throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ putIntLE(out, NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putLegacyString(out, objectName);
+ putLegacyString(out, uploadId);
+
+ putIntLE(out, partNumbers.length);
+ for (int i = 0; i < partNumbers.length; i++) {
+ putIntLE(out, partNumbers[i]);
+ putLegacyString(out, eTags[i]);
+ }
+
+ putLongLE(out, numBytesInParts);
+
+ if (incompleteObjectName == null) {
+ putIntLE(out, 0);
+ } else {
+ putLegacyString(out, incompleteObjectName);
+ }
+ putLongLE(out, incompleteObjectLength);
+
+ return out.toByteArray();
+ }
+
+ private static void putLegacyString(ByteArrayOutputStream out, String
value)
+ throws IOException {
+ byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+ putIntLE(out, bytes.length);
+ out.write(bytes);
+ }
+
+ private static void putIntLE(ByteArrayOutputStream out, int value) throws
IOException {
+ out.write(
+ ByteBuffer.allocate(Integer.BYTES)
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .putInt(value)
+ .array());
+ }
+
+ private static void putLongLE(ByteArrayOutputStream out, long value)
throws IOException {
+ out.write(
+ ByteBuffer.allocate(Long.BYTES)
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .putLong(value)
+ .array());
+ }
}
