eskabetxe commented on code in PR #29422:
URL: https://github.com/apache/flink/pull/29422#discussion_r4422584452
##########
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:
Just to confirm what you mean: the serializer can't drop it, so the skip has
to live in NativeS3Committer.
My idea is to mark recoverables decoded from the hadoop format and skip the
commit only for those when parts.isEmpty() && numBytesInParts == 0. Native's
own behaviour stays the same (it still throws on empty parts).
It can have a gap: if that committable gets checkpointed again before it's
committed, it's written in native format without the mark.
is this what you had in mind, or do you prefer skipping for every empty
recoverable like hadoop does? this will change the native.
--
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]
