[core] Add retry to Consumer.fromPath

herefree · Aug 15, 2024 · 1bb00c0 · 1bb00c0
1 parent e8070eb
commit 1bb00c0
Showing 1 changed file with 19 additions and 4 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java b/paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java
@@ -26,6 +26,7 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.exc.MismatchedInputException;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -58,10 +59,24 @@ public static Consumer fromJson(String json) {
     }
 
     public static Optional<Consumer> fromPath(FileIO fileIO, Path path) {
-        try {
-            return fileIO.readOverwrittenFileUtf8(path).map(Consumer::fromJson);
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
+        int retryNumber = 0;
+        MismatchedInputException exception = null;
+        while (retryNumber++ < 5) {
+            try {
+                return fileIO.readOverwrittenFileUtf8(path).map(Consumer::fromJson);
+            } catch (MismatchedInputException e) {
+                // retry
+                exception = e;
+                try {
+                    Thread.sleep(100);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(ie);
+                }
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
         }
+        throw new UncheckedIOException(exception);
     }
 }