[CELEBORN-1567] Support throw FetchFailedException when Data corrupti…

…on detected

### 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

Closes #2691 from cxzl25/CELEBORN-1567.

Authored-by: sychen <sychen@ctrip.com>
Signed-off-by: Shaoyun Chen <csy@apache.org>

client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java

@@ -27,8 +27,10 @@
 
 import scala.Tuple2;
 
+import com.github.luben.zstd.ZstdException;
 import com.google.common.util.concurrent.Uninterruptibles;
 import io.netty.buffer.ByteBuf;
+import net.jpountz.lz4.LZ4Exception;
 import org.roaringbitmap.RoaringBitmap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -665,15 +667,20 @@ private boolean fillBuffer() throws IOException {
         }
 
         return hasData;
-      } catch (IOException e) {
+      } catch (LZ4Exception | ZstdException | IOException e) {
         logger.error(
             "Failed to fill buffer from chunk. AppShuffleId {}, shuffleId {}, partitionId {}, location {}",
             appShuffleId,
             shuffleId,
             partitionId,
             currentReader.getLocation(),
             e);
-        IOException ioe = e;
+        IOException ioe;
+        if (e instanceof IOException) {
+          ioe = (IOException) e;
+        } else {
+          ioe = new IOException(e);
+        }
         if (exceptionMaker != null) {
           if (shuffleClient.reportShuffleFetchFailure(appShuffleId, shuffleId)) {
             /*
@@ -690,7 +697,7 @@ private boolean fillBuffer() throws IOException {
         throw ioe;
       } catch (Exception e) {
         logger.error(
-            "Failed to read data from chunk. AppShuffleId {}, shuffleId {}, partitionId {}, location {}",
+            "Failed to fill buffer from chunk. AppShuffleId {}, shuffleId {}, partitionId {}, location {}",
             appShuffleId,
             shuffleId,
             partitionId,