[server] Add retry times for ZkSequenceIDCounter and Sender to acquire writer id from zk (#1066)

Author: swuferhong

fluss-client/src/main/java/com/alibaba/fluss/client/write/IdempotenceManager.java

@@ -18,6 +18,7 @@
 
 import com.alibaba.fluss.annotation.Internal;
 import com.alibaba.fluss.annotation.VisibleForTesting;
+import com.alibaba.fluss.exception.AuthorizationException;
 import com.alibaba.fluss.exception.OutOfOrderSequenceException;
 import com.alibaba.fluss.exception.UnknownWriterIdException;
 import com.alibaba.fluss.metadata.PhysicalTablePath;
@@ -27,6 +28,7 @@
 import com.alibaba.fluss.rpc.gateway.TabletServerGateway;
 import com.alibaba.fluss.rpc.messages.InitWriterRequest;
 import com.alibaba.fluss.rpc.protocol.Errors;
+import com.alibaba.fluss.utils.ExceptionUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,7 +37,6 @@
 
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 import static com.alibaba.fluss.record.LogRecordBatch.NO_WRITER_ID;
@@ -53,6 +54,10 @@
 public class IdempotenceManager {
     private static final Logger LOG = LoggerFactory.getLogger(IdempotenceManager.class);
 
+    // retry config for requesting writerId, maybe make it as configurable
+    private static final int RETRY_TIMES = 3;
+    private static final int RETRY_INTERVAL_MS = 100;
+
     private final boolean idempotenceEnabled;
     private final IdempotenceBucketMap idempotenceBucketMap;
     private final int maxInflightRequestsPerBucket;
@@ -283,13 +288,33 @@ synchronized boolean canRetry(WriteBatch batch, TableBucket tableBucket, Errors
         return false;
     }
 
-    void maybeWaitForWriterId(Set<PhysicalTablePath> tablePaths)
-            throws ExecutionException, InterruptedException {
-        if (!isWriterIdValid()) {
-            tabletServerGateway
-                    .initWriter(prepareInitWriterRequest(tablePaths))
-                    .thenAccept(response -> setWriterId(response.getWriterId()))
-                    .get(); // TODO: can optimize into async response handling.
+    void maybeWaitForWriterId(Set<PhysicalTablePath> tablePaths) throws Throwable {
+        if (isWriterIdValid()) {
+            return;
+        }
+
+        int retryCount = 0;
+        while (true) {
+            try {
+                tabletServerGateway
+                        .initWriter(prepareInitWriterRequest(tablePaths))
+                        .thenAccept(response -> setWriterId(response.getWriterId()))
+                        .get(); // TODO: can optimize into async response handling.
+                return;
+            } catch (Exception e) {
+                Throwable t = ExceptionUtils.stripExecutionException(e);
+                if (t instanceof AuthorizationException || retryCount >= RETRY_TIMES) {
+                    throw t;
+                } else {
+                    LOG.warn(
+                            "Failed to init writer id, the retry count: {}, error message: {}",
+                            retryCount,
+                            t.getMessage());
+                    retryCount++;
+                    long delayMs = (long) (RETRY_INTERVAL_MS * Math.pow(2, retryCount));
+                    Thread.sleep(delayMs);
+                }
+            }
         }
     }
 

fluss-client/src/main/java/com/alibaba/fluss/client/write/Sender.java

@@ -178,9 +178,7 @@ public void runOnce() throws Exception {
             // TODO: only request to init writer_id when we have valid target tables
             try {
                 idempotenceManager.maybeWaitForWriterId(targetTables);
-            } catch (Exception e) {
-                Throwable t = ExceptionUtils.stripExecutionException(e);
-
+            } catch (Throwable t) {
                 // TODO: If 'only request to init writer_id when we have valid target tables' have
                 // been down, this if check can be removed.
                 if (!targetTables.isEmpty()) {

fluss-server/src/main/java/com/alibaba/fluss/server/zk/ZkSequenceIDCounter.java

@@ -20,7 +20,7 @@
 import com.alibaba.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;
 import com.alibaba.fluss.shaded.curator5.org.apache.curator.framework.recipes.atomic.AtomicValue;
 import com.alibaba.fluss.shaded.curator5.org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
-import com.alibaba.fluss.shaded.curator5.org.apache.curator.retry.RetryNTimes;
+import com.alibaba.fluss.shaded.curator5.org.apache.curator.retry.BoundedExponentialBackoffRetry;
 
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -30,7 +30,8 @@ public class ZkSequenceIDCounter implements SequenceIDCounter {
 
     // maybe make it as configurable
     private static final int RETRY_TIMES = 10;
-    private static final int RETRY_INTERVAL_MS = 100;
+    private static final int BASE_SLEEP_MS = 100;
+    private static final int MAX_SLEEP_MS = 1000;
 
     private final DistributedAtomicLong sequenceIdCounter;
 
@@ -39,7 +40,8 @@ public ZkSequenceIDCounter(CuratorFramework curatorClient, String sequenceIDPath
                 new DistributedAtomicLong(
                         curatorClient,
                         sequenceIDPath,
-                        new RetryNTimes(RETRY_TIMES, RETRY_INTERVAL_MS));
+                        new BoundedExponentialBackoffRetry(
+                                BASE_SLEEP_MS, MAX_SLEEP_MS, RETRY_TIMES));
     }
 
     /**