[FLINK-30975][filesystems] Upgrade to Hadoop to 3.4.3 and support AWS SDK v2 in flink-filesystem - #27026
[FLINK-30975][filesystems] Upgrade to Hadoop to 3.4.3 and support AWS SDK v2 in flink-filesystem#27026ctrlaltdilj wants to merge 1 commit into
Conversation
ctrlaltdilj
changed the title
ctrlaltdilj
changed the title
7917522 to
2157681
Compare
Samrat002
left a comment
Samrat002
left a comment
There was a problem hiding this comment.
Please ensure
- Code is compiling, and all integration tests are passing properly
- Add proof of code working e2e with S3 filesystem with a sample application
github-actions
Bot
added
the
community-reviewed
|
I am not in a position to approve this as I do not know the area. The title says upgrade hadoop 3.4.x - I am not sure what backports you could do - I assume you would want to deprecate the existing hadoop version and add the new one. |
Samrat002
left a comment
Samrat002
left a comment
There was a problem hiding this comment.
Thank you @ctrlaltdilj for the patch.
I have reviewed the changes at a high level.
Cheers, Samrat
|
This upgrade won't address the concerns from #23844 (comment) - I don't think we can't move forward with this one right now. |
|
@MartijnVisser I was thinking about working through this by creating FlinkS3AFileSystem which provides access to the s3Client, not a huge fan of this(just work around), what do you think? |
|
I have not spend a lot of time fully understanding the hadoop upgrade situation, but before I go into a research rabbithole: |
Because S3 tests only run on merged PRs, because they rely on AWS S3 credentials that aren't in the pipelines because they could get leaked. |
Ok, I see. You can also run them locally by setting these env variables, I believe: @ctrlaltdilj have you validated this PR locally with those env variables set against a real S3 bucket? |
|
@rmetzger @MartijnVisser I believe I had ran this locally a couple months ago, I'll run my test against S3 bucket |
|
@rmetzger @MartijnVisser I have tested locally against a real s3 bucket and the tests are working |
|
let me check if I can simplify these changes and add performance tests |
dc55b6f to
b648d38
Compare
It would be good to first get the CI tests passing |
|
This upgrade is particularly important and useful, given that AWS SDK v1 reached end-of-support in December 2025. |
35e4387 to
2de0941
Compare
|
@rmetzger @MartijnVisser do you think can you enable the CI runs for this PR |
|
I think this has been extensively reviewed, and we'll have some time for testing it from the master branch in the coming months before the next release, so I'm fine merging it. |
|
thanks for the work that has gone into the reviews thus far, happy to tackle any bugs or issues( hopefully none 🤞) that arise. |
|
From our internal experience, we initially started by upgrading Hadoop to 3.x and leveraging SDK v2. As you can see, I have initiated the Jira and shared the first version of the patch. Before we merge this patch, I think the following details need to be added for a better understanding of the behaviour of the upgrade and what we are adding as transitive changes.
@Poorvankbhatia / @rkhachatryan |
rkhachatryan
left a comment
rkhachatryan
left a comment
There was a problem hiding this comment.
I'll try to do a more complete meta review but so far I only have a concern about breaking Presto.
Can you please double check that this PR doesn't introduce class not found exceptions in case when only Presto is used?
Samrat002
left a comment
Samrat002
left a comment
There was a problem hiding this comment.
I have left few comment.
My major concern is this patch upgrade the flink-s3-base and touches flink-s3-presto.
- this can have impact on flink users who uses both connector for a job.
- there are few specific dependencies are removed explicitly which fixes vulnerablities . e.g : https://issues.apache.org/jira/browse/FLINK-21123.
it would be great @ctrlaltdilj if you can share logs and test results of using presto-s3 e2e for a fink job ?
|
taking a look at the comments |
0cc78f8 to
437fc04
Compare
|
Will do some performance tomorrow and report back |
437fc04 to
11e8eae
Compare
718176f to
c30646f
Compare
c30646f to
487154e
Compare
|
All testing was done with Minio: Write (10.28 GB, 5 M × 2 KB records, unthrottled FileSink, 64 MB roll): Read (same fixed 10.28 GB / 156-object dataset, both sides read identical objects): Checkpoint latency (StateMachine-style, 90 checkpoints/run, small state): Checkpoint throughput (~560 MB full hashmap snapshots every 5 s, stats over all >400 MB checkpoints): |
rkhachatryan
left a comment
rkhachatryan
left a comment
There was a problem hiding this comment.
Thanks for updating the PR!
I've left some comments, PTAL.
Apart from that, it looks like configuration is effectively changed upon migration due to inheriting default configs:
The PR's own code changes nothing here — applyHadoopConfigDefaults only pins fs.s3a.input.stream.type=classic; nothing else touches buffers/threads/pools/retries.
But Hadoop's own defaults shifted 3.3.6→3.4.2/3.4.3, and this PR inherits every bit of it silently (verified against Constants.java/AWSClientConfig.java across the actual release tags):
┌─────────────────────────────────────────────────────────┬──────────────────┬───────────────────────────────────────────────────────────┬─────────────────────────────────────────┐
│ Setting │ 3.3.x (SDK v1) │ 3.4.x (SDK v2) │ Change │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ fs.s3a.connection.maximum (HTTP pool size) │ 96 │ 500 │ 5x more connections │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ fs.s3a.threads.max │ 10 │ 96 │ ~10x more threads │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ fs.s3a.attempts.maximum (retry count) │ 10 │ 5 │ halved │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ Retry strategy │ SDK v1 legacy │ SDK v2 RetryMode.ADAPTIVE (client-side throttling-aware │ qualitatively different, not just │
│ │ policy │ rate limiting) │ backoff timing │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ fs.s3a.connection.establish.timeout │ 50s │ 30s │ shorter │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ fs.s3a.connection.acquisition.timeout / │ n/a │ 60s / 5min │ new, SDK v2-only │
│ fs.s3a.connection.ttl │ │ │ │
├─────────────────────────────────────────────────────────┼──────────────────┼───────────────────────────────────────────────────────────┼─────────────────────────────────────────┤
│ Multipart size/threshold, fast-upload-buffer │ 64MB / 128MB / │ unchanged │ not a concern │
│ │ disk │ │ │
└─────────────────────────────────────────────────────────┴──────────────────┴───────────────────────────────────────────────────────────┴─────────────────────────────────────────┘
I don't think we should preserve the config strictly as-is, but such changes should be justified and documented.
| // S3A keeps zero-byte directory markers since Hadoop 3.4 | ||
| // (fs.s3a.directory.marker.retention defaults to "keep"); only actual | ||
| // ApplicationResultStore entries are relevant for the assertions. | ||
| .filter(summary -> !summary.getKey().endsWith("/")) | ||
| .collect(Collectors.toList()); |
There was a problem hiding this comment.
Why can't we keep the current behavior by configuring fs.s3a.directory.marker.retention?
There was a problem hiding this comment.
good call, fixed, so we now set s3.directory.marker.retention: delete in the test config instead and dropped the listing filter
| final CompletedPart completedPart = | ||
| CompletedPart.builder().partNumber(partNumber).eTag(result.eTag()).build(); | ||
| future.complete(completedPart); |
There was a problem hiding this comment.
With SDK v2, this becomes incorrect if user sets s.s3a.checksum.generation=true. In that case, S3 will require checksum per part and fail the request because the checksum is missing.
There was a problem hiding this comment.
thanks for calling that out, I fixed it by copying every checksum field from the UploadPartResponse into the CompletedPart (centralized in a new S3PartChecksum enum so the set can't drift from the SDK). That also required bumping S3RecoverableSerializer to v2, recoverable persisted at checkpoints only stored partNumber and eTag, so a commit after recovery would have failed the same way. v1 state deserializes as before (it predates checksum generation, so checksum-less is correct)
|
|
||
| @Override | ||
| public AwsCredentials resolveCredentials() throws SdkException { | ||
| S3SessionCredentials credentials = AbstractS3DelegationTokenReceiver.getCredentials(); |
There was a problem hiding this comment.
This is fragile: this code (HadoopDynamicTemporaryAWSCredentialsProvider) has only sdk v1 on its classpath; however, AbstractS3DelegationTokenReceiver imports sdk v2 code.
If a AbstractS3DelegationTokenReceiver / S3SessionCredentials method that uses sdk v2 classes is called (in the future); that will result in class not found exception.
I'm not sure it's worth fixing it though given that we'll have v2 (and native) S3.
There was a problem hiding this comment.
Addressed via S3SessionCredentials which is a plain POJO with no SDK imports, and conversion to v1/v2 types happens only inside each plugin's own credentials provider. Let me know if you prefer to handle this differently
|
addressing the comments today |
|
Cool, thank you! |
6409732 to
b99b991
Compare
|
@rkhachatryan thanks for your review, I have made some of the fixes you pointed out, let me know what you think |
b99b991 to
eb0a7ab
Compare
10 participants
What is the purpose of the change
Hadoop 3.4.X introduced 2500+ bug fixes and support for AWS SDK V2. Note AWS SDK V1 is EOL 12/31/2025.
One challenge is maintaining support for presto which is using AWS SDK V1 and is not updated just yet, While Hadoop 3.4.X has support for AWS SDK V1 there are a couple wrapper classes need to support the SDK changes and maintain support for Presto's AWS SDK V1 and support Hadoop's AWS SDK V2 upgrade.
Brief change log
Verifying this change
This change is already covered by existing tests and adds new test coverage:
Existing tests:
- All existing unit tests for flink-s3-fs-hadoop (HadoopS3FileSystemTest, HadoopS3FileSystemsSchemesTest) pass with Hadoop 3.4.2 and AWS
SDK V2
- All existing unit tests for flink-s3-fs-presto continue to pass with AWS SDK V1
- Integration tests (HAJobRunOnHadoopS3FileSystemITCase, S5CmdOnHadoopS3FileSystemITCase) verify S3 functionality with new SDK
New/updated tests:
- Converted
S3FileSystemMinioTestandPrestoS3FileSystemMinioTestE2E tests to JUnit framework- Both tests verify write, read, and delete operations against MinIO (S3-compatible storage)
- Tests confirm that both Hadoop (SDK V2) and Presto (SDK V1) filesystems work correctly
Manual verification:
- Verified AWS SDK V1 is completely removed from flink-s3-fs-hadoop JAR (0 classes from
com.amazonaws.*)- Verified AWS SDK V1 remains in flink-s3-fs-presto JAR as expected
- Confirmed multipart upload operations work correctly with new HadoopS3AccessHelper
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation