[PR] [WORKER] Preserve partially-committed partitions on CommitFiles timeout [celeborn]

["shlomitubul (via GitHub)" <gi...@apache.org>]

shlomitubul opened a new pull request, #3705:
URL: https://github.com/apache/celeborn/pull/3705

   ## What changes were proposed in this pull request?
   
   When the worker-side `celeborn.worker.commitFiles.timeout` fires and `future.cancel(true)` interrupts the per-partition commit tasks, `Controller`'s BiFunction has two issues that amplify data loss unnecessarily:
   
   1. **The response is built with `List.empty.asJava`** for both `committedPrimaryIds` and `committedReplicaIds`, even though those concurrent sets may have been partially populated by tasks that finished before the timer fired. All successful commit work is silently thrown away.
   
   2. **`context.reply()` is never called on the error path**, so the originating commit RPC sits unanswered until the driver's `celeborn.client.rpc.commitFiles.askTimeout` expires. The worker has already determined the outcome — there's no reason to make the driver wait.
   
   This PR:
   
   - Builds the response from the actual state of `committedPrimaryIds` / `committedReplicaIds` / `failedPrimaryIds` / `failedReplicaIds` / `committedPrimaryStorageInfos` / `committedReplicaStorageInfos` / `committedMapIdBitMap` / `partitionSizeList`.
   - Returns `StatusCode.PARTIAL_SUCCESS` when any partition committed before cancellation, with the populated lists. `CommitHandler` on the client already treats `PARTIAL_SUCCESS` as a terminal, non-retry status (alongside `SUCCESS`, `SHUFFLE_UNREGISTERED`, `REQUEST_FAILED`, `WORKER_EXCLUDED`, `COMMIT_FILE_EXCEPTION`), so no client-side change is required.
   - Preserves the existing `StatusCode.COMMIT_FILE_EXCEPTION` response when nothing committed.
   - Calls `context.reply(response)` so the driver's RPC ask receives the verdict immediately instead of timing out.
   
   ## Why are the changes needed?
   
   In production, we hit `celeborn.worker.commitFiles.timeout` periodically on heavy shuffles where the per-worker partition count makes the close() work exceed the timeout. When that happens today:
   
   - The driver receives no reply and times out at `commitFiles.askTimeout`, logging `Cannot receive any reply ... in 300000 milliseconds`.
   - Even partitions whose close() ran to completion before the timer fired are reported as not committed (because of the empty lists).
   - The driver marks the *entire* shuffle as data-lost via `dataLostShuffleSet.add(shuffleId)`, even when most partitions succeeded.
   - Every reducer for that shuffle hits `SHUFFLE_DATA_LOST` → `FetchFailedException` → DAGScheduler retries the *whole* map stage.
   
   With this change, the driver receives a definitive `PARTIAL_SUCCESS` reply with the actual committed/failed split. The data-lost set is populated based on the partitions that genuinely didn't make it. Reducers for the partitions that *did* commit can fetch them normally.
   
   ## Does this PR introduce any user-facing change?
   
   No client API or wire format changes. `StatusCode.PARTIAL_SUCCESS` is already part of the `CommitFilesResponse` protocol and is already handled by the driver-side `CommitHandler.parallelCommitFiles` retry loop. The user-visible effect is reduced blast radius when a worker's commit times out under heavy load.
   
   ## How was this patch tested?
   
   - `./build/mvn -DskipTests install` builds the full tree cleanly (verified locally with Java 17 / Maven 3.9).
   - The change preserves the existing `COMMIT_FILE_EXCEPTION` path byte-for-byte for the "nothing committed" case; the new branch reuses the exact same construction pattern as the success-path `reply()` method on lines 622-654 (`committedPrimaryStorageAndDiskHintList`, `committedReplicaStorageAndDiskHintList`, `committedMapIdBitMapList`, etc.).
   - All required imports (`jArrayList`, `jHashMap`, `RoaringBitmap`, `StorageInfo`) are already in scope at the top of `Controller.scala`.
   - Unit/integration tests for the partial-success-on-timeout path are TBD; happy to add a test for `CommitFilesResponse` field population on cancellation if the reviewers can point me at the right test class to extend.


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [WORKER] Preserve partially-committed partitions on CommitFiles timeout [celeborn]

["shlomitubul (via GitHub)" <gi...@apache.org>]

shlomitubul commented on PR #3705:
URL: https://github.com/apache/celeborn/pull/3705#issuecomment-4556640610

   Refiling against branch-0.6 (0.6.x maintenance line).


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [WORKER] Preserve partially-committed partitions on CommitFiles timeout [celeborn]

["shlomitubul (via GitHub)" <gi...@apache.org>]

shlomitubul closed pull request #3705: [WORKER] Preserve partially-committed partitions on CommitFiles timeout
URL: https://github.com/apache/celeborn/pull/3705


-- 
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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org