[zaynt4606]

JIRA: https://issues.apache.org/jira/browse/CELEBORN-2508

What changes were proposed in this pull request?

Retain a partial-read integrity validation failure for the lifetime of the logical shuffle. Reject subsequent completion reports and shuffle generation requests, and decline conversion of that failure into a Spark fetch failure. Clear the failure when the logical shuffle is unregistered.

Why are the changes needed?

AQE can split one reducer partition across several tasks. Earlier tasks may return corrupted data before the last task reports the aggregate CRC mismatch. Spark can then regenerate the shuffle but rerun only the final missing result task, retaining the earlier corrupted results. In a controlled Spark reproduction, the query returned all 16,000 rows, including one corrupted row, after detecting the mismatch and rerunning the stage.

A new physical shuffle does not invalidate those already accepted results. Fail the query after a detected partial-read integrity failure until recovery can guarantee that every affected result is recomputed. Ordinary fetch failures and whole-partition integrity failures retain stage rerun.

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

A detected AQE aggregate integrity failure fails the query instead of attempting a stage rerun that may preserve corrupted results. No new configurations, RPC messages, or worker changes are required. Healthy reads do not add RPCs or per-record processing.

How was this patch tested?

• Community main: 60 client lifecycle and commit tests passed, including six new RPC-dispatch regressions for legacy and indexed partial reads, a previously allocated physical generation, ordinary fetch recovery, unrelated shuffles, and failure cleanup.
• The equivalent lifecycle fix was also built on a 0.5 client and tested with EMR Spark 3.5.3 on one ECS master and two real remote workers. A production SparkShuffleManager probe used local[4,2], controlled payload corruption, AQE target 128 KiB, no compression, and no replication. The baseline reproduced a successful query with one corrupt row. The fixed client rejected the query with both values of the skew fetch-abort option; an uncorrupted AQE query returned all 16,000 expected rows. A separate whole-partition corruption control recovered through stage rerun with correct results.
• The ECS workload is not a distributed YARN validation of the community-main binary.
• Ran the complete documented dev/reformat workflow.

[yew1eb]

LGTM