gaborgsomogyi opened a new pull request, #29429:
URL: https://github.com/apache/flink/pull/29429
## What is the purpose of the change
When the native S3 sink checkpoints a file it is still writing, it has to
save the data that is buffered but too small to upload yet (S3 multipart
uploads need parts of at least 5 MiB). It saves that leftover (the "tail" in
the code) as a small extra object, so the sink can carry on after a failure.
Until now that object always went to a fixed folder at the top of the
bucket, no matter where the data is written:
```
writing s3://my-bucket/events/dt=2026-10-07/part-0
before s3://my-bucket/.incomplete/<uploadId>/<uuid>
after s3://my-bucket/events/dt=2026-10-07/.incomplete/<uploadId>/<uuid>
```
That is a problem when the credentials are limited to the folder the data
goes to (IAM or session policies scoped to a prefix): the sink may write its
data but not the extra object, so the checkpoint cannot store it. With this
change the extra object lives next to the data, in a hidden `.incomplete`
folder that Flink's file source skips, so whatever permission covers the data
covers it too. The Hadoop based S3 file system already does the same.
Nothing else changes: checkpoints remember the exact key of the extra object
and recovery just reads that key, so checkpoints and savepoints created before
this change still restore, and the checkpoint format is untouched.
## Brief change log
- The key of the extra object now starts with the folder of the file being
written (`NativeS3RecoverableFsDataOutputStream#persist`); files at the bucket
root behave exactly as before
- New `NativeS3RecoverableWriterTest`, one serializer test, and the shared
in-memory S3 fake moved to `FakeNativeS3Operations`
- `NativeS3RecoverableWriterRecoveryITCase` updated to the new location
## Verifying this change
This change added tests and can be verified as follows:
- `persistStoresTailUnderTargetDirectory` checks the new location; it
fails without the change and passes with it
- `recoverResumesFromTailAtStoredKey` and
`cleanupDeletesOnlyTailAtStoredKey` check that recovery and cleanup work with
the old and the new location, so upgrading with existing checkpoints is safe
- `testVersion1BytesWithTailKeyAtBucketRoot` checks that the stored format
did not change
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no (the format is unchanged)
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, only where the S3
sink keeps its small extra object during a checkpoint; restoring old
checkpoints works as before
- The S3 file system connector: yes
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code, Sonnet 5.5)
Generated-by: Claude Code (Sonnet 5.5)
--
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]
