[saurabhd336]

Summary:
Problem

The LifecycleManager treats the first MapperEnd for a map index as the attempt reducers read (ReducePartitionCommitHandler.finishMapperAttempt is first-wins), and sends StageEnd as soon as every map index has one. The writer calls mapperEnd from inside write(), before Spark has decided whether the task succeeded.

When an executor is preempted while running with
spark.sql.files.ignoreCorruptFiles=true, reads from the closed FileSystem fail silently, the input iterator ends early and the task still calls mapperEnd with truncated output. Spark marks the task failed and runs a new attempt, but Celeborn has already picked the truncated attempt, and the retry is blocked from writing its data:

• if the bad attempt was the last mapper, StageEnd already committed the files and the retry's pushes are dropped as stage-ended;
• otherwise handleRevive reports the map as ended, and the retry's client adds it to mapperEndMap and stops pushing;
• mid-stage HARD_SPLIT CommitFiles copy the bad attempt to the workers, which then answer the retry's pushes with MAP_ENDED.

Reducers end up reading the incomplete attempt.

Fix

Behind celeborn.client.spark.shuffle.mapperEndConfirmation.enabled (default false), a MapperEnd only counts once Spark's MapOutputTracker has registered that exact task attempt for the map index:

• PbMapperEnd gets optional int64 taskAttemptId = 6 (MapperEnd defaults it to -1 when unset). ShuffleClient gains a mapperEnd overload carrying it; the spark-3 sort and hash writers pass taskContext.taskAttemptId(), which is also what Spark records as MapStatus.mapId.
• LifecycleManager replies SUCCESS to MapperEnd immediately (the task can't finish, and Spark can't register it, until the call returns) and parks the request in the new MapperEndConfirmer.
• MapperEndConfirmer polls every celeborn.client.spark.shuffle.mapperEndConfirmation.interval (100ms) on a daemon thread. When the registered task attempt for a map matches a pending entry, it runs the existing finishMapperAttempt/StageEnd logic (moved to confirmMapperEnd) exactly once, and drops the other pending attempts of that map.
• Until confirmed, a map is not ended: revives and pushes from a retry keep working, and mid-stage CommitFiles send -1 for it so workers answer HARD_SPLIT rather than MAP_ENDED. No worker changes are needed, since the worker's mapper attempts are a copy of the LifecycleManager's array.
• GetReducerFileGroup runs one synchronous check, so the first reducer does not wait for the next poll.
• unregisterShuffle drains pending entries before forcing StageEnd: maps without a Spark status fall back to first-wins; if Spark accepted an attempt that never reported MapperEnd, the shuffle is marked data lost (skipped for outdated Celeborn shuffle ids after a stage rerun).
• SparkShuffleManager registers SparkUtils.getRegisteredMapTaskIds, which reads MapOutputTrackerMaster.shuffleStatuses, as the resolver. Without a resolver (Spark 2, Flink, MR) or without taskAttemptId (older clients) the legacy first-wins path is used unchanged.

Metrics: mapper_end_rejected_count, mapper_end_fallback_count, mapper_end_confirm_latency, mapper_end_pending, and cause accepted_attempt_not_seen on stage_end_failure_count.

Tests

• MapperEndConfirmerSuite: no confirmation before registration, confirm the registered attempt and drop others, exactly-once under concurrent checks, stage-end cleanup, fallback paths, background thread.
• MapperEndConfirmationSuite (spark-it, mini cluster): a truncated attempt that reports MapperEnd followed by a full retry; with confirmation the reducer reads only the retry, without it reads the truncated data.
• UtilsSuite: MapperEnd pb round trip with and without taskAttemptId.

Also adds .claude/ to .gitignore and includes a log line for the selected shuffle write mode in SparkShuffleManager.

Co-Authored-By: Claude Opus 5.5 noreply@anthropic.com

Reviewers: #ldap_hadoop-compute-drogon!

Tags: #has_java

Differential Revision: https://code.uberinternal.com/D29771827

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

How was this patch tested?

[saurabhd336]

TODO: Looks like with "spark.shuffle.useOldFetchProtocol" enabled, this might be wrong since the mapId is the partitionId and not taskAttemptId. Will add a todo and fix in in a followup