eskabetxe commented on code in PR #29422:
URL: https://github.com/apache/flink/pull/29422#discussion_r4422631511
##########
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:
done
--
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]
