HADOOP-19569. S3A: stream write/close fails badly once FS is closed #7700
+1,568
−879
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Executors in hadoop-common to - pick up shutdown of inner executor and shut themselves down. - semaphore executor to decrement counters in this process so that queue state is updated This stops callers being able to submit work when the inner executor has shut down. S3A code - StoreImpl to IllegalStateException on method invocation whene the service isn't running. Some methods are kept open as they do seem needed. - WriteOperationHelper callbacks to raise IllegalStateException when invoked when FS is closed. This is complex. TODO: - WriteOperationHelper MUST make all calls to the FS through its callback interface, rather than given a ref to S3AFS. This makes it easy to identify and lock down the methods. - What is the correct exception to raise in write/close() failures? IOE or illegal state?
Took the opportunity to move all of WriteOperationHelper to calling store, not FS, which then required all multipart IO to go there too. Next stage of store design: multipart IO is its own service underneath the store, so keeping the size of Store interface and Impl under control.
steveloughran
force-pushed
the
s3/HADOOP-19569-stream-write-fs-close
branch
from
cf32b54 to
fea3e89
Compare
Continue migration * move nearly all MPU ops out of s3afs, resulting, so shaving code out of that * Multipart service retrieved and invoked as appropriate * StoreImpl stores a map of ServiceName -> service. with a lookupService() method in S3AStore interface, it's possible to retrieve services through the API just by knowing their name and type * registering all current services this way * fixing up the mock tests to work
* rate limiting api to hadoop-aws * semaphored delegating excutor Itest to hadoop common unit test
|
💔 -1 overall
This message was automatically generated. |
There was a problem hiding this comment.
Pull Request Overview
This PR addresses HADOOP-19569 by enhancing the S3A filesystem’s behavior when closed, ensuring that write/close operations fail gracefully, and improves internal service registration and executor shutdown behavior. Key changes include enforcing FS state checks via added checkRunning() calls, refactoring helper and callback methods (e.g. renaming getWriteOperationHelper to createWriteOperationHelperWithinActiveSpan and replacing direct FS calls with getStore() invocations), and cleaning up legacy utilities such as MultipartUtils.
Reviewed Changes
Copilot reviewed 32 out of 32 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/MockS3AFileSystem.java | Adjusted WriteOperationHelper construction with updated callback parameters and method renaming. |
| hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMiscOperations.java hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AClientSideEncryption.java |
Updated test calls to use getStore() for invoking putObjectDirect. |
| hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3AStoreImpl.java hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AStore.java |
Inserted checkRunning() calls across API methods and updated service lookup/register methods. |
| hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/WriteOperationHelper.java | Refactored to use a new “callbacks” field and improved null validations with requireNonNull. |
| hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java | Exposed getStore() publicly and updated multipart upload handling to leverage the new MultipartIOService. |
| hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestBlockingThreadPoolExecutorService.java hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/SemaphoredDelegatingExecutor.java hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/BlockingThreadPoolExecutorService.java |
Enhanced executor service tests and added rejection-handling logic to ensure proper shutdown on task rejection. |
Comments suppressed due to low confidence (3)
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/WriteOperationHelper.java:127
- [nitpick] Update the Javadoc in WriteOperationHelper to reflect the change from using the variable name 'writeOperationHelperCallbacks' to 'callbacks', ensuring that the documentation clearly explains its role and usage.
private final WriteOperationHelperCallbacks callbacks;
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AStore.java:290
- The completeMultipartUpload method has been removed from the S3AStore interface. Ensure that all external consumers have been updated accordingly or consider adding a deprecation warning to ease the transition.
/* Removed completeMultipartUpload method */
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java:1357
- [nitpick] Making the getStore() method public exposes internal implementation details. Confirm that this exposure is intended, or consider providing a more restricted accessor if external access is not required.
public S3AStore getStore() {
| @@ -130,21 +130,20 @@ public static BlockingThreadPoolExecutorService newInstance( | |||
| slower than enqueueing. */ | |||
| final BlockingQueue<Runnable> workQueue = | |||
| new LinkedBlockingQueue<>(waitingTasks + activeTasks); | |||
| final InnerExecutorRejection rejection = new InnerExecutorRejection(); | |||
There was a problem hiding this comment.
[nitpick] The InnerExecutorRejection handler now shuts down the service upon rejection. Consider enhancing the error handling logic or adding more detailed documentation to explain the shutdown behavior in case of task rejection.
Copilot uses AI. Check for mistakes.
| private static final int BLOCKING_THRESHOLD_MSEC = 50; | ||
|
|
||
| private static final Integer SOME_VALUE = 1337; | ||
|
|
||
| private static BlockingThreadPoolExecutorService tpe; | ||
| private BlockingThreadPoolExecutorService tpe; |
There was a problem hiding this comment.
[nitpick] Since the thread pool executor is now an instance variable with setup/teardown methods, ensure that each test properly initializes and destroys the executor to avoid interference between tests.
Copilot uses AI. Check for mistakes.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
HADOOP-19569.
Executors in hadoop-common to
S3A code
This is complex.
TODO:
How was this patch tested?
New ITests which close the FS while simple and multipart writes are in progress.
S3 london.
For code changes:
LICENSE,LICENSE-binary,NOTICE-binaryfiles?