You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/08/15 04:04:44 UTC
(paimon) branch master updated: [core] Add retry to Consumer.fromPath
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1bb00c01a [core] Add retry to Consumer.fromPath
1bb00c01a is described below
commit 1bb00c01a8e3d5693ebfe05ce03b66c283ac21f4
Author: Jingsong <ji...@gmail.com>
AuthorDate: Thu Aug 15 12:00:57 2024 +0800
[core] Add retry to Consumer.fromPath
---
.../java/org/apache/paimon/consumer/Consumer.java | 23 ++++++++++++++++++----
1 file changed, 19 insertions(+), 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
index ffa63ece3..bad0e2509 100644
--- 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.JsonCre
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 class Consumer {
}
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);
}
}