[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 <[email protected]> Signed-off-by: Shaoyun Chen <[email protected]> (cherry picked from commit b8f275d) Signed-off-by: Shaoyun Chen <[email protected]>
apache · Aug 20, 2024 · 08716da · 08716da
1 parent b5ac49f
commit 08716da
Showing 1 changed file with 10 additions and 3 deletions.
diff --git a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java b/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;
@@ -664,15 +666,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)) {
             /*
@@ -689,7 +696,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,