venkata91 opened a new pull request, #3776: URL: https://github.com/apache/celeborn/pull/3776
### What changes were proposed in this pull request? `ShuffleClientImpl.updateFileGroup` issued the blocking `GetReducerFileGroup` RPC inside `reduceFileGroupsMap.compute(shuffleId, ...)`. `ConcurrentHashMap.compute` holds the per-key bin lock for the whole remapping function, so the RPC round-trip to the `LifecycleManager` ran while holding that lock. This moves the RPC out from under the bin lock: a lock-free `get()` fast path so cache hits take no lock, and first-time loads serialize on a dedicated per-shuffle monitor (`fileGroupLoadLocks`, cleared in `cleanupShuffle`) with a double-check so still only one RPC is issued per shuffle. Caching semantics are unchanged: a cached tuple whose `_1()` is `null` (failed load) still triggers a reload, exactly as the old `compute()` branch did. `FlinkShuffleClientImpl` already overrides `updateFileGroup` and is unaffected; the Spark plugin uses the base `ShuffleClientImpl` directly (`ShuffleClient.get()` -> `new ShuffleClientImpl(...)`, via `CelebornShuffleReader`), so it was exposed. This is the standalone convoy fix extracted from #3687 (closed), independent of the partition-range metadata optimization in #3745 (CELEBORN-2370) that supersedes it. Credit to @sunchao for #3687. ### Why are the changes needed? At high reduce parallelism, the first reduce task takes the bin lock and fires `GetReducerFileGroup` (seconds under load for a large shuffle); every other reduce task for the same `shuffleId` then blocks on that bin lock inside `compute`, even though they are cache hits. The result is a thundering-herd convoy: thousands of reduce tasks BLOCKED on `ConcurrentHashMap.compute` with no fetch activity and no fetch failures, so the stage appears frozen with nothing on fetch dashboards. Because a failed load is cached as a `null` tuple and `CelebornShuffleReader` retries until stage end, a slow RPC re-forms the convoy each retry, turning a transient stall into a multi-minute wedge. Observed in production (tens of thousands of reduce partitions, ~900 executors): a stage sat idle 20+ min; the driver thread dump showed reduce threads BLOCKED in `updateFileGroup` -> `ConcurrentHashMap.compute`. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New unit test `ShuffleClientSuiteJ#testUpdateReducerFileGroupConcurrentLoadIssuesSingleRpc`: 16 threads call `updateFileGroup` for one shuffle against a mock RPC that blocks 500ms; asserts all callers complete and exactly one RPC is issued. Existing `ShuffleClientSuiteJ` passes 14/14 (incl. interrupt, non-fetch-failure, timeout paths). `build/mvn -pl client -am test -Dtest=ShuffleClientSuiteJ` and `spotless:check` green. -- 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]
