[server] Add retriable authentication exception. (apache#845)

Author: loserwang1024

LICENSE

@@ -311,6 +311,7 @@ Apache Kafka
 ./fluss-common/src/main/java/com/alibaba/fluss/utils/crc/PureJavaCrc32C.java
 ./fluss-common/src/main/java/com/alibaba/fluss/utils/log/ByteBufferUnmapper.java
 ./fluss-common/src/main/java/com/alibaba/fluss/utils/log/FairBucketStatusMap.java
+./fluss-common/src/main/java/com/alibaba/fluss/utils/ExponentialBackoff.java
 ./fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/statemachine/ReplicaStateMachine.java
 ./fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
 ./fluss-server/src/main/java/com/alibaba/fluss/server/log/AbstractIndex.java

fluss-common/src/main/java/com/alibaba/fluss/exception/RetriableAuthenticationException.java

@@ -0,0 +1,28 @@
+/*
+ *  Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ *  Licensed 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 com.alibaba.fluss.exception;
+
+/**
+ * This exception is thrown if authentication fails with a retriable error.
+ *
+ * @since 0.7
+ */
+public class RetriableAuthenticationException extends AuthenticationException {
+    public RetriableAuthenticationException(String message) {
+        super(message);
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/security/auth/ClientAuthenticator.java

@@ -28,6 +28,9 @@ public interface ClientAuthenticator {
     /** The protocol name of the authenticator, which will send in the AuthenticateRequest. */
     String protocol();
 
+    /** Initialize the authenticator. */
+    default void initialize(AuthenticateContext context) {}
+
     /**
      * * Generates the initial token or calculates a token based on the server's challenge, then
      * sends it back to the server. This method sets the client authentication status as complete if
@@ -75,4 +78,7 @@ public interface ClientAuthenticator {
 
     /** Checks if the authentication from client side is completed. */
     boolean isCompleted();
+
+    /** The context of the authentication process. */
+    interface AuthenticateContext {}
 }

fluss-common/src/main/java/com/alibaba/fluss/security/auth/ServerAuthenticator.java

@@ -30,6 +30,9 @@ public interface ServerAuthenticator {
 
     String protocol();
 
+    /** Initialize the authenticator. */
+    default void initialize(AuthenticateContext context) {}
+
     /**
      * * Generates the challenge based on the client's token, then sends it back to the client. This
      * method sets the server authentication status as complete if the authentication succeeds.
@@ -79,4 +82,7 @@ public interface ServerAuthenticator {
      * complete).
      */
     FlussPrincipal createPrincipal();
+
+    /** The context of the authentication process. */
+    interface AuthenticateContext {}
 }

fluss-common/src/main/java/com/alibaba/fluss/utils/ExponentialBackoff.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.utils;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.concurrent.ThreadLocalRandom;
+
+/* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A utility class for computing exponential backoff values, commonly used for retry logic,
+ * reconnection attempts, and timeout management.
+ *
+ * <p>The backoff interval increases exponentially with each attempt, following the formula:
+ *
+ * <pre>
+ * Backoff(attempts) = random(1 - jitter, 1 + jitter) * initialInterval * multiplier^attempts
+ * </pre>
+ *
+ * <p>If {@code maxInterval} is less than {@code initialInterval}, a constant backoff of {@code
+ * maxInterval} will be applied. The jitter factor ensures randomness to avoid thundering herd
+ * problems, but will never cause the result to exceed the configured maximum interval.
+ *
+ * <p>Instances of this class are thread-safe and can be shared across multiple threads.
+ */
+@ThreadSafe
+public class ExponentialBackoff {
+    private final long initialInterval;
+    private final int multiplier;
+    private final long maxInterval;
+    private final double jitter;
+    private final double expMax;
+
+    public ExponentialBackoff(
+            long initialInterval, int multiplier, long maxInterval, double jitter) {
+        this.initialInterval = Math.min(maxInterval, initialInterval);
+        this.multiplier = multiplier;
+        this.maxInterval = maxInterval;
+        this.jitter = jitter;
+        this.expMax =
+                maxInterval > initialInterval
+                        ? Math.log(maxInterval / (double) Math.max(initialInterval, 1))
+                                / Math.log(multiplier)
+                        : 0;
+    }
+
+    public long backoff(long attempts) {
+        if (expMax == 0) {
+            return initialInterval;
+        }
+        double exp = Math.min(attempts, this.expMax);
+        double term = initialInterval * Math.pow(multiplier, exp);
+        double randomFactor =
+                jitter < Double.MIN_NORMAL
+                        ? 1.0
+                        : ThreadLocalRandom.current().nextDouble(1 - jitter, 1 + jitter);
+        long backoffValue = (long) (randomFactor * term);
+        return Math.min(backoffValue, maxInterval);
+    }
+
+    @Override
+    public String toString() {
+        return "ExponentialBackoff{"
+                + "multiplier="
+                + multiplier
+                + ", expMax="
+                + expMax
+                + ", initialInterval="
+                + initialInterval
+                + ", jitter="
+                + jitter
+                + '}';
+    }
+}

fluss-common/src/test/java/com/alibaba/fluss/utils/ExponentialBackoffTest.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.utils;
+
+import org.assertj.core.data.Percentage;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link ExponentialBackoff}. */
+public class ExponentialBackoffTest {
+    @Test
+    public void testExponentialBackoff() {
+        long initialValue = 100;
+        int ratio = 2;
+        long backoffMax = 2000;
+        double jitter = 0.2;
+        ExponentialBackoff exponentialBackoff =
+                new ExponentialBackoff(initialValue, ratio, backoffMax, jitter);
+
+        for (int i = 0; i <= 100; i++) {
+            for (int attempts = 0; attempts <= 10; attempts++) {
+                if (attempts <= 4) {
+                    assertThat(1.0 * exponentialBackoff.backoff(attempts))
+                            .isCloseTo(
+                                    initialValue * Math.pow(ratio, attempts),
+                                    Percentage.withPercentage(jitter * 100));
+                } else {
+                    assertThat(exponentialBackoff.backoff(attempts) <= backoffMax * (1 + jitter))
+                            .isTrue();
+                }
+            }
+        }
+    }
+
+    @Test
+    public void testExponentialBackoffWithoutJitter() {
+        ExponentialBackoff exponentialBackoff = new ExponentialBackoff(100, 2, 400, 0.0);
+        assertThat(exponentialBackoff.backoff(0)).isEqualTo(100);
+        assertThat(exponentialBackoff.backoff(1)).isEqualTo(200);
+        assertThat(exponentialBackoff.backoff(2)).isEqualTo(400);
+        assertThat(exponentialBackoff.backoff(3)).isEqualTo(400);
+    }
+}

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/client/ServerConnection.java

@@ -20,6 +20,7 @@
 import com.alibaba.fluss.exception.DisconnectException;
 import com.alibaba.fluss.exception.FlussRuntimeException;
 import com.alibaba.fluss.exception.NetworkException;
+import com.alibaba.fluss.exception.RetriableAuthenticationException;
 import com.alibaba.fluss.rpc.messages.ApiMessage;
 import com.alibaba.fluss.rpc.messages.ApiVersionsRequest;
 import com.alibaba.fluss.rpc.messages.ApiVersionsResponse;
@@ -38,6 +39,7 @@
 import com.alibaba.fluss.shaded.netty4.io.netty.channel.Channel;
 import com.alibaba.fluss.shaded.netty4.io.netty.channel.ChannelFuture;
 import com.alibaba.fluss.shaded.netty4.io.netty.channel.ChannelFutureListener;
+import com.alibaba.fluss.utils.ExponentialBackoff;
 import com.alibaba.fluss.utils.MapUtils;
 
 import org.slf4j.Logger;
@@ -52,6 +54,7 @@
 import java.util.ArrayDeque;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 /** Connection to a Netty server used by the {@link NettyClient}. */
@@ -66,6 +69,7 @@ final class ServerConnection {
     private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
     private final ConnectionMetricGroup connectionMetricGroup;
     private final ClientAuthenticator authenticator;
+    private final ExponentialBackoff backoff;
 
     private final Object lock = new Object();
 
@@ -87,6 +91,9 @@ final class ServerConnection {
     @GuardedBy("lock")
     private ServerApiVersions serverApiVersions;
 
+    @GuardedBy("lock")
+    private int retryAuthCount = 0;
+
     ServerConnection(
             Bootstrap bootstrap,
             ServerNode node,
@@ -99,6 +106,7 @@ final class ServerConnection {
                 .connect(node.host(), node.port())
                 .addListener((ChannelFutureListener) this::establishConnection);
         this.authenticator = authenticator;
+        this.backoff = new ExponentialBackoff(100L, 2, 5000L, 0.2);
     }
 
     public ServerNode getServerNode() {
@@ -366,13 +374,18 @@ private void handleApiVersionsResponse(ApiMessage response, Throwable cause) {
         synchronized (lock) {
             serverApiVersions =
                     new ServerApiVersions(((ApiVersionsResponse) response).getApiVersionsList());
-            LOG.debug("Begin to authenticate with protocol {}", authenticator.protocol());
             // send initial token
-            sendAuthenticate(new byte[0]);
+            sendInitialToken();
         }
     }
 
-    private void sendAuthenticate(byte[] challenge) {
+    private void sendInitialToken() {
+        authenticator.initialize(new DefaultAuthenticateContext());
+        LOG.debug("Begin to authenticate with protocol {}", authenticator.protocol());
+        sendAuthenticateRequest(new byte[0]);
+    }
+
+    private void sendAuthenticateRequest(byte[] challenge) {
         try {
             if (!authenticator.isCompleted()) {
                 byte[] token = authenticator.authenticate(challenge);
@@ -403,19 +416,28 @@ private void sendAuthenticate(byte[] challenge) {
     }
 
     private void handleAuthenticateResponse(ApiMessage response, Throwable cause) {
-        if (cause != null) {
-            close(cause);
-            return;
-        }
-        if (!(response instanceof AuthenticateResponse)) {
-            close(new IllegalStateException("Unexpected response type " + response.getClass()));
-            return;
-        }
-
         synchronized (lock) {
+            if (cause != null) {
+                if (cause instanceof RetriableAuthenticationException) {
+                    LOG.warn("Authentication failed, retrying {} times", retryAuthCount, cause);
+                    channel.eventLoop()
+                            .schedule(
+                                    this::sendInitialToken,
+                                    backoff.backoff(retryAuthCount++),
+                                    TimeUnit.MILLISECONDS);
+                } else {
+                    close(cause);
+                }
+                return;
+            }
+            if (!(response instanceof AuthenticateResponse)) {
+                close(new IllegalStateException("Unexpected response type " + response.getClass()));
+                return;
+            }
+
             AuthenticateResponse authenticateResponse = (AuthenticateResponse) response;
             if (authenticateResponse.hasChallenge()) {
-                sendAuthenticate(((AuthenticateResponse) response).getChallenge());
+                sendAuthenticateRequest(((AuthenticateResponse) response).getChallenge());
             } else if (authenticator.isCompleted()) {
                 switchState(ConnectionState.READY);
             } else {
@@ -521,4 +543,7 @@ ByteBuf toByteBuf(ByteBufAllocator allocator) {
             return MessageCodec.encodeRequest(allocator, apiKey, apiVersion, requestId, request);
         }
     }
+
+    private static class DefaultAuthenticateContext
+            implements ClientAuthenticator.AuthenticateContext {}
 }