Samrat002 commented on code in PR #29422:
URL: https://github.com/apache/flink/pull/29422#discussion_r4397431277
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java:
##########
@@ -148,4 +152,320 @@ void testVersion1BytesWithTailKeyAtBucketRoot() throws
IOException {
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();
+ }
+
+ @Test
+ void testCurrentFormatNotMistakenForLegacyFormat() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+
+ List<NativeS3Recoverable.PartETag> parts = new ArrayList<>();
+ parts.add(new NativeS3Recoverable.PartETag(1, "etag1"));
+ NativeS3Recoverable original =
+ new NativeS3Recoverable("test-object-key", "test-upload-id",
parts, 12345L);
+
+ byte[] serialized = serializer.serialize(original);
+ NativeS3Recoverable deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(deserialized.getObjectName()).isEqualTo(original.getObjectName());
+ assertThat(deserialized.uploadId()).isEqualTo(original.uploadId());
+
assertThat(deserialized.numBytesInParts()).isEqualTo(original.numBytesInParts());
+ assertThat(deserialized.parts()).hasSize(1);
+ }
+
+ /**
+ * Golden test against 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), 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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200"
+ + "0000010000000800"
+ + "0000657461672d6f"
+ + "6e65020000000800"
+ + "0000657461672d74"
+ + "776f393000000000"
+ + "000011000000696e"
+ + "636f6d706c657465"
+ + "2d6b65792dc3a400"
+ + "04000000000000"));
+
+
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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200");
+ 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 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() {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ assertThatThrownBy(
+ () -> {
+ putIntLE(out,
NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putIntLE(out, -1);
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ out.toByteArray());
+ })
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid field length");
+ }
+
+ @Test
+ void testDeserializeLegacyWithHugeLengthThrowsIOException() {
Review Comment:
Same as above
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java:
##########
@@ -148,4 +152,320 @@ void testVersion1BytesWithTailKeyAtBucketRoot() throws
IOException {
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();
+ }
+
+ @Test
+ void testCurrentFormatNotMistakenForLegacyFormat() throws IOException {
Review Comment:
This is the same as the existing round-trip tests and doesn't test the
format check. Either drop it, or build current-format bytes starting with 0x32
0x14 (UTF length 12820) and assert they don't go down the legacy path.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -108,6 +133,12 @@ public NativeS3Recoverable deserialize(int version, byte[]
serialized) throws IO
if (version != CURRENT_VERSION) {
throw new IOException("Unsupported version: " + version);
}
+ if (isLegacyFormat(serialized)) {
Review Comment:
Hadoop's S3Committer.commit() / commitAfterRecovery() do nothing when
totalLength == 0. NativeS3Committer.commit() throws "Cannot commit empty
multipart upload" when parts is empty. So a Hadoop committable with no parts
will now decode, then fail on every restore. Could we either skip the commit
for parts.isEmpty() && numBytesInParts == 0, or list this as a known limitation
in the docs note?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +166,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}).
+ *
+ * <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).",
+ 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
Review Comment:
nit: this check can't fail, since isLegacyFormat just checked the same
thing. Fine to keep for parity.
Just noting it.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +166,119 @@ public NativeS3Recoverable deserialize(int version,
byte[] serialized) throws IO
incompleteObjectName,
incompleteObjectLength);
}
+
+ /**
+ * Checks whether the given serialized state was written by the {@code
S3RecoverableSerializer}
Review Comment:
Overwhelming desciption
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +166,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}).
+ *
+ * <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.
+ */
Review Comment:
Overwhelming desciption
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java:
##########
@@ -148,4 +152,320 @@ void testVersion1BytesWithTailKeyAtBucketRoot() throws
IOException {
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();
+ }
+
+ @Test
+ void testCurrentFormatNotMistakenForLegacyFormat() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+
+ List<NativeS3Recoverable.PartETag> parts = new ArrayList<>();
+ parts.add(new NativeS3Recoverable.PartETag(1, "etag1"));
+ NativeS3Recoverable original =
+ new NativeS3Recoverable("test-object-key", "test-upload-id",
parts, 12345L);
+
+ byte[] serialized = serializer.serialize(original);
+ NativeS3Recoverable deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(deserialized.getObjectName()).isEqualTo(original.getObjectName());
+ assertThat(deserialized.uploadId()).isEqualTo(original.uploadId());
+
assertThat(deserialized.numBytesInParts()).isEqualTo(original.numBytesInParts());
+ assertThat(deserialized.parts()).hasSize(1);
+ }
+
+ /**
+ * Golden test against 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), 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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200"
+ + "0000010000000800"
+ + "0000657461672d6f"
+ + "6e65020000000800"
+ + "0000657461672d74"
+ + "776f393000000000"
+ + "000011000000696e"
+ + "636f6d706c657465"
+ + "2d6b65792dc3a400"
+ + "04000000000000"));
+
+
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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200");
+ 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 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() {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ assertThatThrownBy(
+ () -> {
+ putIntLE(out,
NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putIntLE(out, -1);
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ out.toByteArray());
+ })
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid field length");
+ }
+
+ @Test
+ void testDeserializeLegacyWithHugeLengthThrowsIOException() {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ assertThatThrownBy(
+ () -> {
+ putIntLE(out,
NativeS3RecoverableSerializer.LEGACY_MAGIC_NUMBER);
+ putIntLE(out, Integer.MAX_VALUE);
+ NativeS3RecoverableSerializer.INSTANCE.deserialize(
+
NativeS3RecoverableSerializer.INSTANCE.getVersion(),
+ out.toByteArray());
+ })
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid field length");
+ }
+
+ @Test
+ void testDeserializeLegacyWithHugePartCountThrowsIOException() {
Review Comment:
Same as above
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java:
##########
@@ -148,4 +152,320 @@ void testVersion1BytesWithTailKeyAtBucketRoot() throws
IOException {
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();
+ }
+
+ @Test
+ void testCurrentFormatNotMistakenForLegacyFormat() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+
+ List<NativeS3Recoverable.PartETag> parts = new ArrayList<>();
+ parts.add(new NativeS3Recoverable.PartETag(1, "etag1"));
+ NativeS3Recoverable original =
+ new NativeS3Recoverable("test-object-key", "test-upload-id",
parts, 12345L);
+
+ byte[] serialized = serializer.serialize(original);
+ NativeS3Recoverable deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(deserialized.getObjectName()).isEqualTo(original.getObjectName());
+ assertThat(deserialized.uploadId()).isEqualTo(original.uploadId());
+
assertThat(deserialized.numBytesInParts()).isEqualTo(original.numBytesInParts());
+ assertThat(deserialized.parts()).hasSize(1);
+ }
+
+ /**
+ * Golden test against 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), 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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200"
+ + "0000010000000800"
+ + "0000657461672d6f"
+ + "6e65020000000800"
+ + "0000657461672d74"
+ + "776f393000000000"
+ + "000011000000696e"
+ + "636f6d706c657465"
+ + "2d6b65792dc3a400"
+ + "04000000000000"));
+
+
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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200");
+ 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 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() {
Review Comment:
[NIT] setup runs inside assertThatThrownBy, so an IOException from setup
would also satisfy the IOException assertion. Could the bytes be built before
the assertion, so the lambda only calls deserialize?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +166,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}).
+ *
+ * <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 "
Review Comment:
nit: the other errors use "Corrupt legacy S3 recoverable state: ...". Could
this use the same prefix, so callers and tests can match one message?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
@@ -135,4 +166,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}).
+ *
+ * <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()
Review Comment:
nit, optional: comparing the first 4 bytes directly (0x32, 0x14, 0x76, 0x98)
would skip allocating a buffer. Not on a hot path, so feel free to ignore.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializer.java:
##########
Review Comment:
This comment sits between the field and the constructor and isn't attached
to anything. The description also says the base decoder must stay untouched for
backports. Could we drop this change ?
##########
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.
+
Review Comment:
Why this is required ?
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java:
##########
@@ -148,4 +152,320 @@ void testVersion1BytesWithTailKeyAtBucketRoot() throws
IOException {
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();
+ }
+
+ @Test
+ void testCurrentFormatNotMistakenForLegacyFormat() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+
+ List<NativeS3Recoverable.PartETag> parts = new ArrayList<>();
+ parts.add(new NativeS3Recoverable.PartETag(1, "etag1"));
+ NativeS3Recoverable original =
+ new NativeS3Recoverable("test-object-key", "test-upload-id",
parts, 12345L);
+
+ byte[] serialized = serializer.serialize(original);
+ NativeS3Recoverable deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(deserialized.getObjectName()).isEqualTo(original.getObjectName());
+ assertThat(deserialized.uploadId()).isEqualTo(original.uploadId());
+
assertThat(deserialized.numBytesInParts()).isEqualTo(original.numBytesInParts());
+ assertThat(deserialized.parts()).hasSize(1);
+ }
+
+ /**
+ * Golden test against 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), 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(
+ "3214769816000000"
+ + "6469722f74657374"
+ + "2d6b65792dc3a4e6"
+ + "97a5f09f94910c00"
+ + "000075706c6f6164"
+ + "2dc3a42d69640200"
+ + "0000010000000800"
+ + "0000657461672d6f"
+ + "6e65020000000800"
+ + "0000657461672d74"
+ + "776f393000000000"
+ + "000011000000696e"
+ + "636f6d706c657465"
+ + "2d6b65792dc3a400"
+ + "04000000000000"));
+
+
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() {
Review Comment:
This input fails in readBoundedLength (upload-id length 12 > remaining), so
the BufferUnderflowException / IllegalArgumentException wrapping in
deserializeLegacy is never tested. Could you add:
- the golden bytes with the last byte cut off (underflow while reading the
final long)
- legacy state with an incomplete name but incompleteObjectLength <= 0 (the
constructor's checkArgument should come back as an IOException)
--
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]
