Skip to content

[CELEBORN-2252] Skip writing data from other attempts #3590

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
taowenjun wants to merge 2 commits into
base: main
Choose a base branch
from
+25 −1
Open

[CELEBORN-2252] Skip writing data from other attempts #3590

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
03274f8
[CELEBORN-2252] Skip writing data from other attempts after one of th…
taowenjun Jan 21, 2026
28b9e8a
[CELEBORN-2252] Skip writing data from other attempts after one of th…
taowenjun Jan 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 25 additions & 1 deletion worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1522,7 +1522,31 @@ class PushDataHandler(val workerSource: WorkerSource) extends BaseMessageHandler
shuffleKey: String,
index: Int): Unit = {
try {
fileWriter.write(body)
val (endedAttempt, toWrite, curMapId, curMapAttempt) =
Copy link
Contributor

@RexXiong RexXiong Jan 27, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Celeborn already check that otherMapAttempt is completed before executing writeData to avoid unnecessary writes. Why is it necessary to perform an additional check within writeData to skip data writing?

Copy link
Contributor Author

@taowenjun taowenjun Feb 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As communicated, I have added logs to prove it.

Copy link
Contributor

@RexXiong RexXiong Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, you're correct. Previously, the mapper end check was performed only when the partition location was null during the handling of PushData/PushMergeData. I believe we can enhance this by extending the check to the mapper level, regardless of whether the partition location is null.

if (shuffleMapperAttempts.containsKey(shuffleKey)) {
val (mapId, attemptId) = getMapAttempt(body)
if (mapId >= shuffleMapperAttempts.get(shuffleKey).length()) {
throw new ArrayIndexOutOfBoundsException(s"MapId $mapId of shuffleKey " +
s"$shuffleKey out of array index bounds when accessing shuffleMapperAttempts.")
}
val endedAttemptId = shuffleMapperAttempts.get(shuffleKey).get(mapId)
Copy link

Copilot AI Jan 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Potential ArrayIndexOutOfBoundsException when accessing shuffleMapperAttempts. If mapId is greater than or equal to the length of the AtomicIntegerArray, this will throw an exception. Consider adding a bounds check before accessing the array, similar to how other parts of the codebase handle this.

Copilot uses AI. Check for mistakes.
val toWriteAttempt =
if (endedAttemptId == -1) {
// Map task has not completed yet, always write
true
} else {
// Only write if the current attempt matches the ended attempt
attemptId == endedAttemptId
}
(endedAttemptId, toWriteAttempt, mapId, attemptId)
} else (-1, true, -1, -1)
if (toWrite) {
fileWriter.write(body)
} else {
fileWriter.decrementPendingWrites()
logInfo(
s"Skipping data from map $curMapId attempt $curMapAttempt for shuffle $shuffleKey " +
s"because map already completed with attempt $endedAttempt") }
result(index) = StatusCode.SUCCESS
} catch {
case e: Throwable =>
Expand Down
Loading