sunchao opened a new pull request, #3784: URL: https://github.com/apache/celeborn/pull/3784
## Why are the changes needed? Workers currently wait synchronously for on-demand reduce partition sorting during both individual and batched open-stream requests. The wait polls every 50 ms and can occupy a fetch RPC thread for the full configured sort timeout, which defaults to 220 seconds. Under concurrent AQE or skewed reduce reads, these waits can exhaust fetch threads and delay unrelated shuffle reads. [CELEBORN-2065](https://issues.apache.org/jira/browse/CELEBORN-2065) tracks this problem. Previous upstream proposals in #3593 and #3652 identified the same fetch-thread starvation but closed before an implementation was merged. ## What changes were proposed in this PR? - Add an asynchronous sorted-file lookup while retaining the existing synchronous sorter API. - Coalesce requests for the same in-flight sort using completion futures scoped by both shuffle key and file identity. - Resolve sorted indexes on a bounded, dedicated pool so unrelated reads can progress independently without blocking the worker's shared RPC reply executor. - Make individual, legacy, and batched reduce-stream opens asynchronous while preserving batch response order, wire compatibility, existing storage backends, and per-file failures. - Propagate sort failures, configured wait timeouts, expired-shuffle cleanup, and worker shutdown to pending readers. - Finalize sorted output streams before making files visible, and preserve failed-sort ownership until its output handles have closed. - Keep per-shuffle index-cache bookkeeping thread-safe and bound graceful shutdown across sorter and resolver executors. No RPC protocol changes or new configuration settings are introduced. ## How was this PR tested? - Applied repository-wide formatting: ```bash build/mvn --no-transfer-progress -DskipTests spotless:apply ``` - Ran the affected worker Java suites and relevant Scala storage suite: ```bash build/mvn --no-transfer-progress -pl worker -am \ -Dtest=FetchHandlerSuiteJ,DiskPartitionFilesSorterSuiteJ,DiskReducePartitionDataWriterSuiteJ,MemoryReducePartitionDataWriterSuiteJ \ -DwildcardSuites=org.apache.celeborn.service.deploy.worker.storage.PartitionMetaHandlerSuite \ -DfailIfNoTests=false -Dsurefire.failIfNoSpecifiedTests=false test ``` - The affected suite run passed 34 Java tests and 3 Scala tests. After the final concurrency refinements, the timeout, retry, and fetch regression subset was rerun and passed 10 Java tests plus the same 3 Scala tests. - Added regression coverage for nonblocking individual and batched reads, ordered batch responses, asynchronous sort failures, shared waiters, cross-shuffle key collisions, exact-key cleanup, timeout retries, independent resolver progress, failed-sort retry ownership, and requests arriving after sorter shutdown. -- 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]
