modify based on cr

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/security/auth/ClientAuthenticator.java

@@ -18,7 +18,6 @@
 
 import com.alibaba.fluss.annotation.PublicEvolving;
 import com.alibaba.fluss.exception.AuthenticationException;
-import com.alibaba.fluss.shaded.netty4.io.netty.channel.Channel;
 
 import javax.annotation.Nullable;
 
@@ -81,7 +80,5 @@ default void initialize(AuthenticateContext context) {}
     boolean isCompleted();
 
     /** The context of the authentication process. */
-    interface AuthenticateContext {
-        Channel channel();
-    }
+    interface AuthenticateContext {}
 }

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

@@ -19,7 +19,6 @@
 import com.alibaba.fluss.annotation.PublicEvolving;
 import com.alibaba.fluss.exception.AuthenticationException;
 import com.alibaba.fluss.security.acl.FlussPrincipal;
-import com.alibaba.fluss.shaded.netty4.io.netty.channel.Channel;
 
 /**
  * Authenticator for server side.
@@ -85,7 +84,5 @@ default void initialize(AuthenticateContext context) {}
     FlussPrincipal createPrincipal();
 
     /** The context of the authentication process. */
-    interface AuthenticateContext {
-        Channel channel();
-    }
+    interface AuthenticateContext {}
 }

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

@@ -416,25 +416,25 @@ private void sendAuthenticateRequest(byte[] challenge) {
     }
 
     private void handleAuthenticateResponse(ApiMessage response, Throwable cause) {
-        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);
+        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;
             }
-            return;
-        }
-        if (!(response instanceof AuthenticateResponse)) {
-            close(new IllegalStateException("Unexpected response type " + response.getClass()));
-            return;
-        }
 
-        synchronized (lock) {
             AuthenticateResponse authenticateResponse = (AuthenticateResponse) response;
             if (authenticateResponse.hasChallenge()) {
                 sendAuthenticateRequest(((AuthenticateResponse) response).getChallenge());
@@ -544,10 +544,6 @@ ByteBuf toByteBuf(ByteBufAllocator allocator) {
         }
     }
 
-    private class DefaultAuthenticateContext implements ClientAuthenticator.AuthenticateContext {
-        @Override
-        public Channel channel() {
-            return channel;
-        }
-    }
+    private static class DefaultAuthenticateContext
+            implements ClientAuthenticator.AuthenticateContext {}
 }

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/server/NettyServerHandler.java

@@ -33,7 +33,6 @@
 import com.alibaba.fluss.security.auth.ServerAuthenticator;
 import com.alibaba.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
 import com.alibaba.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator;
-import com.alibaba.fluss.shaded.netty4.io.netty.channel.Channel;
 import com.alibaba.fluss.shaded.netty4.io.netty.channel.ChannelFutureListener;
 import com.alibaba.fluss.shaded.netty4.io.netty.channel.ChannelHandlerContext;
 import com.alibaba.fluss.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter;
@@ -366,10 +365,6 @@ public boolean isAuthenticating() {
         }
     }
 
-    private class DefaultAuthenticateContext implements ServerAuthenticator.AuthenticateContext {
-        @Override
-        public Channel channel() {
-            return ctx.channel();
-        }
-    }
+    private static class DefaultAuthenticateContext
+            implements ServerAuthenticator.AuthenticateContext {}
 }