[core] Fix retry in Consumer and SnapshotManager

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/utils/RetryUtils.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Optional;
+import java.util.function.Function;
+
+/** Utils for retry operations. */
+public class RetryUtils {
+
+    public static <I, R> Optional<R> retry(
+            SupplierWithIOException<I> inputSupplier, Function<I, R> retryFunc) {
+        int retryNumber = 0;
+        Exception exception = null;
+        while (retryNumber++ < 10) {
+            I input;
+            try {
+                input = inputSupplier.get();
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
+            if (input == null) {
+                return Optional.empty();
+            }
+            try {
+                return Optional.ofNullable(retryFunc.apply(input));
+            } catch (Exception e) {
+                // retry
+                exception = e;
+                try {
+                    Thread.sleep(200);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(ie);
+                }
+            }
+        }
+        throw new RuntimeException("Retry fail after 10 times.", exception);
+    }
+}

paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java

@@ -21,15 +21,13 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.paimon.utils.RetryUtils;
 
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 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;
 import java.util.Optional;
 
 /** Consumer which contains next snapshot. */
@@ -59,24 +57,7 @@ public static Consumer fromJson(String json) {
     }
 
     public static Optional<Consumer> fromPath(FileIO fileIO, Path path) {
-        int retryNumber = 0;
-        MismatchedInputException exception = null;
-        while (retryNumber++ < 10) {
-            try {
-                return fileIO.readOverwrittenFileUtf8(path).map(Consumer::fromJson);
-            } catch (MismatchedInputException e) {
-                // retry
-                exception = e;
-                try {
-                    Thread.sleep(1_000);
-                } catch (InterruptedException ie) {
-                    Thread.currentThread().interrupt();
-                    throw new RuntimeException(ie);
-                }
-            } catch (IOException e) {
-                throw new UncheckedIOException(e);
-            }
-        }
-        throw new UncheckedIOException(exception);
+        return RetryUtils.retry(
+                () -> fileIO.readOverwrittenFileUtf8(path).orElse(null), Consumer::fromJson);
     }
 }

paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java

@@ -24,7 +24,6 @@
 import org.apache.paimon.table.Instant;
 
 import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.exc.MismatchedInputException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -771,26 +770,7 @@ public static Snapshot fromPath(FileIO fileIO, Path path) {
     }
 
     public static Snapshot tryFromPath(FileIO fileIO, Path path) throws FileNotFoundException {
-        int retryNumber = 0;
-        MismatchedInputException exception = null;
-        while (retryNumber++ < 10) {
-            try {
-                return Snapshot.fromJson(fileIO.readFileUtf8(path));
-            } catch (MismatchedInputException e) {
-                // retry
-                exception = e;
-                try {
-                    Thread.sleep(1_000);
-                } catch (InterruptedException ie) {
-                    Thread.currentThread().interrupt();
-                    throw new RuntimeException(ie);
-                }
-            } catch (FileNotFoundException e) {
-                throw e;
-            } catch (IOException e) {
-                throw new RuntimeException("Fails to read snapshot from path " + path, e);
-            }
-        }
-        throw new UncheckedIOException(exception);
+        return RetryUtils.retry(() -> fileIO.readFileUtf8(path), Snapshot::fromJson)
+                .orElseThrow(() -> new RuntimeException("Cannot read snapshot from " + path));
     }
 }

paimon-core/src/test/java/org/apache/paimon/consumer/ConsumerManagerTest.java

@@ -25,12 +25,15 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import java.io.File;
+import java.io.IOException;
 import java.nio.file.Path;
 import java.time.LocalDateTime;
 import java.util.Optional;
 import java.util.OptionalLong;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link ConsumerManager}. */
 public class ConsumerManagerTest {
@@ -53,6 +56,15 @@ public void before() {
                         "branch1");
     }
 
+    @Test
+    public void testRetry() throws IOException {
+        File file = new File(tempDir.toFile(), "consumer/consumer-id1");
+        file.getParentFile().mkdir();
+        file.createNewFile();
+        assertThatThrownBy(() -> manager.consumer("id1"))
+                .hasMessageContaining("Retry fail after 10 times");
+    }
+
     @Test
     public void test() {
         Optional<Consumer> consumer = manager.consumer("id1");