[CELEBORN-1567] Support throw FetchFailedException when Data corruption detected

[cxzl25]

What changes were proposed in this pull request?

Why are the changes needed?

#2655 (review)

Does this PR introduce any user-facing change?

No

How was this patch tested?

GA

[mridulm]

Looks good to me.

Btw, support for checksum/validation of data would a good feature to add IMO ... there were corner cases where this helped catch issues in spark (instead of relying on compression/deserialization failing ... which need not always happen).

[RexXiong]

LGTM

[cfmcgrady]

I have a question: should we retry fetching another replication before throwing a FetchFailedException when the conf celeborn.client.push.replicate.enabled is set to true?

[cxzl25]

I have a question: should we retry fetching another replication before throwing a FetchFailedException when the conf celeborn.client.push.replicate.enabled is set to true?

This is not necessarily safe, because the Task may have read part of the data, so it is safer to retry the Task. This is how Spark handles it.

[cxzl25]

support for checksum/validation of data would a good feature

It looks like we've already done this.

celeborn/client/src/main/java/org/apache/celeborn/client/compress/ZstdDecompressor.java

Lines 69 to 72 in 0ee3c3a

	if ((int) checksum.getValue() != check) {
	logger.error("Checksum not equal! expected: {}, actual: {}.", check, checksum.getValue());
	return -1;
	}

celeborn/client/src/main/java/org/apache/celeborn/client/compress/Lz4Decompressor.java

Lines 82 to 85 in 0ee3c3a

	if ((int) checksum.getValue() != check) {
	logger.error("Checksum not equal! expected: {}, actual: {}.", check, checksum.getValue());
	return -1;
	}

[FMX]

LGTM.

[cfmcgrady]

LGTM

[cfmcgrady]

Thank you, merging to main(v0.6.0)/branch-0.5(v0.5.2)/branch-0.4(v0.4.3).