fix crc & slow request detector

Author: platinumhamburg

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -463,7 +463,7 @@ public class ConfigOptions {
     public static final ConfigOption<Duration> RPC_SERVER_SLOW_REQUEST_THRESHOLD =
             key("rpc.server.slow-request.threshold")
                     .durationType()
-                    .defaultValue(Duration.ofSeconds(30))
+                    .defaultValue(Duration.ofSeconds(60))
                     .withDescription(
                             "The threshold for identifying slow RPC requests on the server side. "
                                     + "When a request takes longer than this threshold to process, "

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java

@@ -333,29 +333,18 @@ private CompletableFuture<ApiMessage> doSend(
 
             int currentRequestId = requestCount++;
 
-            // Create timeout timer task
-            RequestTimeoutTask timeoutTask =
-                    new RequestTimeoutTask(currentRequestId, DEFAULT_REQUEST_TIMEOUT_MS);
-
+            // Create InflightRequest without timeout task initially
+            // Timeout will be started only after the request is successfully sent to network
             InflightRequest inflight =
                     new InflightRequest(
-                            apiKey.id,
-                            version,
-                            currentRequestId,
-                            rawRequest,
-                            responseFuture,
-                            timeoutTask);
+                            apiKey.id, version, currentRequestId, rawRequest, responseFuture, null);
             inflightRequests.put(inflight.requestId, inflight);
             ByteBuf byteBuf;
             try {
                 byteBuf = inflight.toByteBuf(channel.alloc());
             } catch (Exception e) {
                 LOG.error("Failed to encode request for '{}'.", ApiKeys.forId(inflight.apiKey), e);
                 inflightRequests.remove(inflight.requestId);
-                // Cancel timeout task
-                if (inflight.timeoutTask != null) {
-                    inflight.timeoutTask.cancel();
-                }
                 responseFuture.completeExceptionally(
                         new FlussRuntimeException(
                                 String.format(
@@ -367,19 +356,27 @@ private CompletableFuture<ApiMessage> doSend(
 
             connectionMetrics.updateMetricsBeforeSendRequest(apiKey, rawRequest.totalSize());
 
-            // Add timeout task to timer
-            timer.add(timeoutTask);
+            // Send request to network
             channel.writeAndFlush(byteBuf)
                     .addListener(
                             (ChannelFutureListener)
                                     future -> {
-                                        if (!future.isSuccess()) {
+                                        if (future.isSuccess()) {
+                                            // Start timeout timer only after successfully sent to
+                                            // network
+                                            // This ensures timeout only measures network round-trip
+                                            // + server processing time,
+                                            // excluding client-side queuing time (e.g., semaphore
+                                            // waiting)
+                                            RequestTimeoutTask timeoutTask =
+                                                    new RequestTimeoutTask(
+                                                            currentRequestId,
+                                                            DEFAULT_REQUEST_TIMEOUT_MS);
+                                            inflight.timeoutTask = timeoutTask;
+                                            timer.add(timeoutTask);
+                                        } else {
                                             connectionMetrics.updateMetricsAfterGetResponse(
                                                     apiKey, inflight.requestStartTime, 0);
-                                            // Cancel timeout task
-                                            if (inflight.timeoutTask != null) {
-                                                inflight.timeoutTask.cancel();
-                                            }
                                             Throwable cause = future.cause();
                                             if (cause instanceof IOException) {
                                                 // when server close the channel, the cause will be
@@ -569,7 +566,7 @@ private static class InflightRequest {
         final ApiMessage request;
         final long requestStartTime;
         final CompletableFuture<ApiMessage> responseFuture;
-        final TimerTask timeoutTask;
+        volatile TimerTask timeoutTask;
 
         InflightRequest(
                 short apiKey,

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java

@@ -71,6 +71,12 @@ public List<String> listenerNames() {
     @Override
     public ChannelHandler createChannelHandler(
             RequestChannel[] requestChannels, String listenerName) {
+        boolean slowRequestMonitoringEnabled =
+                conf.get(ConfigOptions.RPC_SERVER_SLOW_REQUEST_MONITORING_ENABLED);
+        long slowRequestThresholdMs =
+                conf.get(ConfigOptions.RPC_SERVER_SLOW_REQUEST_THRESHOLD).toMillis();
+        boolean dumpStack = conf.get(ConfigOptions.RPC_SERVER_SLOW_REQUEST_DUMP_STACK);
+
         return new ServerChannelInitializer(
                 requestChannels,
                 apiManager,
@@ -81,7 +87,11 @@ public ChannelHandler createChannelHandler(
                 Optional.ofNullable(
                                 AuthenticationFactory.loadServerAuthenticatorSuppliers(conf)
                                         .get(listenerName))
-                        .orElse(PlainTextAuthenticationPlugin.PlainTextServerAuthenticator::new));
+                        .orElse(PlainTextAuthenticationPlugin.PlainTextServerAuthenticator::new),
+                slowRequestMonitoringEnabled,
+                slowRequestThresholdMs,
+                dumpStack,
+                timer);
     }
 
     @Override

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequest.java

@@ -23,10 +23,13 @@
 import org.apache.fluss.security.acl.FlussPrincipal;
 import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
 import org.apache.fluss.shaded.netty4.io.netty.util.ReferenceCountUtil;
+import org.apache.fluss.timer.TimerTask;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.net.InetAddress;
 import java.util.concurrent.CompletableFuture;
 
@@ -54,6 +57,9 @@ public final class FlussRequest implements RpcRequest {
     private volatile long requestCompletedTimeMs;
     private volatile boolean cancelled = false;
 
+    // slow request detector, started when request is received
+    @Nullable private volatile TimerTask slowRequestDetector;
+
     public FlussRequest(
             short apiKey,
             short apiVersion,
@@ -167,6 +173,15 @@ public boolean isInternal() {
         return isInternal;
     }
 
+    public void setSlowRequestDetector(TimerTask slowRequestDetector) {
+        this.slowRequestDetector = slowRequestDetector;
+    }
+
+    @Nullable
+    public TimerTask getSlowRequestDetector() {
+        return slowRequestDetector;
+    }
+
     @Override
     public String toString() {
         return "FlussRequest{"

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java

@@ -39,10 +39,6 @@ public class FlussRequestHandler implements RequestHandler<FlussRequest> {
     private static final Logger LOG = LoggerFactory.getLogger(FlussRequestHandler.class);
 
     private final RpcGatewayService service;
-    private final boolean slowRequestMonitoringEnabled;
-    private final long slowRequestThresholdMs;
-    private final boolean dumpStack;
-    @Nullable private final Timer timer;
 
     public FlussRequestHandler(
             RpcGatewayService service,
@@ -51,10 +47,8 @@ public FlussRequestHandler(
             boolean dumpStack,
             @Nullable Timer timer) {
         this.service = service;
-        this.slowRequestMonitoringEnabled = slowRequestMonitoringEnabled;
-        this.slowRequestThresholdMs = slowRequestThresholdMs;
-        this.dumpStack = dumpStack;
-        this.timer = timer;
+        // These parameters are kept for backward compatibility but not used anymore
+        // The slow request monitoring is now done in NettyServerHandler
     }
 
     @Override
@@ -68,21 +62,14 @@ public void processRequest(FlussRequest request) {
         ApiMethod api = request.getApiMethod();
         ApiMessage message = request.getMessage();
 
-        // Schedule slow request detection if enabled
-        TimerTask slowRequestDetector = null;
-        if (slowRequestMonitoringEnabled && timer != null && slowRequestThresholdMs > 0) {
-            slowRequestDetector =
-                    new SlowRequestDetector(
-                            request,
-                            Thread.currentThread(),
-                            slowRequestThresholdMs,
-                            dumpStack,
-                            slowRequestThresholdMs);
-            timer.add(slowRequestDetector);
+        // Update the processing thread in the existing slow request detector
+        // The detector was already started when the request was received (in NettyServerHandler)
+        // Now we update it with the actual processing thread
+        TimerTask detectorTask = request.getSlowRequestDetector();
+        if (detectorTask instanceof SlowRequestDetector) {
+            ((SlowRequestDetector) detectorTask).setProcessingThread(Thread.currentThread());
         }
 
-        final TimerTask detectorTask = slowRequestDetector;
-
         try {
             service.setCurrentSession(
                     new Session(

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java

@@ -39,12 +39,16 @@
 import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter;
 import org.apache.fluss.shaded.netty4.io.netty.handler.timeout.IdleState;
 import org.apache.fluss.shaded.netty4.io.netty.handler.timeout.IdleStateEvent;
+import org.apache.fluss.timer.Timer;
+import org.apache.fluss.timer.TimerTask;
 import org.apache.fluss.utils.ExceptionUtils;
 import org.apache.fluss.utils.IOUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.ArrayDeque;
@@ -71,6 +75,12 @@ public final class NettyServerHandler extends ChannelInboundHandlerAdapter {
 
     private final ServerAuthenticator authenticator;
 
+    // Slow request monitoring configuration
+    private final boolean slowRequestMonitoringEnabled;
+    private final long slowRequestThresholdMs;
+    private final boolean dumpStack;
+    @Nullable private final Timer timer;
+
     private volatile ConnectionState state;
     private volatile boolean initialized = false;
 
@@ -80,13 +90,21 @@ public NettyServerHandler(
             String listenerName,
             boolean isInternal,
             RequestsMetrics requestsMetrics,
-            ServerAuthenticator authenticator) {
+            ServerAuthenticator authenticator,
+            boolean slowRequestMonitoringEnabled,
+            long slowRequestThresholdMs,
+            boolean dumpStack,
+            @Nullable Timer timer) {
         this.requestChannel = requestChannel;
         this.apiManager = apiManager;
         this.listenerName = listenerName;
         this.isInternal = isInternal;
         this.requestsMetrics = requestsMetrics;
         this.authenticator = authenticator;
+        this.slowRequestMonitoringEnabled = slowRequestMonitoringEnabled;
+        this.slowRequestThresholdMs = slowRequestThresholdMs;
+        this.dumpStack = dumpStack;
+        this.timer = timer;
         this.state = ConnectionState.START;
     }
 
@@ -133,6 +151,20 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                             ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress(),
                             future);
 
+            // Schedule slow request detection immediately when request is received
+            // This ensures we monitor the total server-side time including queue waiting
+            if (slowRequestMonitoringEnabled && timer != null && slowRequestThresholdMs > 0) {
+                TimerTask slowRequestDetector =
+                        new SlowRequestDetector(
+                                request,
+                                null,
+                                slowRequestThresholdMs,
+                                dumpStack,
+                                slowRequestThresholdMs);
+                request.setSlowRequestDetector(slowRequestDetector);
+                timer.add(slowRequestDetector);
+            }
+
             future.whenCompleteAsync((r, t) -> sendResponse(ctx, request), ctx.executor());
             if (apiKey == ApiKeys.AUTHENTICATE.id
                     || (state.isAuthenticating() && apiKey != ApiKeys.API_VERSIONS.id)) {

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/ServerChannelInitializer.java

@@ -23,11 +23,14 @@
 import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandlerContext;
 import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelInitializer;
 import org.apache.fluss.shaded.netty4.io.netty.channel.socket.SocketChannel;
+import org.apache.fluss.timer.Timer;
 import org.apache.fluss.utils.MathUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.util.function.Supplier;
 
 /**
@@ -43,6 +46,10 @@ final class ServerChannelInitializer extends NettyChannelInitializer {
     private final boolean isInternal;
     private final RequestsMetrics requestsMetrics;
     private final Supplier<ServerAuthenticator> authenticatorSupplier;
+    private final boolean slowRequestMonitoringEnabled;
+    private final long slowRequestThresholdMs;
+    private final boolean dumpStack;
+    @Nullable private final Timer timer;
 
     public ServerChannelInitializer(
             RequestChannel[] requestChannels,
@@ -51,14 +58,22 @@ public ServerChannelInitializer(
             boolean isInternal,
             RequestsMetrics requestsMetrics,
             long maxIdleTimeSeconds,
-            Supplier<ServerAuthenticator> authenticatorSupplier) {
+            Supplier<ServerAuthenticator> authenticatorSupplier,
+            boolean slowRequestMonitoringEnabled,
+            long slowRequestThresholdMs,
+            boolean dumpStack,
+            @Nullable Timer timer) {
         super(maxIdleTimeSeconds);
         this.requestChannels = requestChannels;
         this.apiManager = apiManager;
         this.endpointListenerName = endpointListenerName;
         this.isInternal = isInternal;
         this.requestsMetrics = requestsMetrics;
         this.authenticatorSupplier = authenticatorSupplier;
+        this.slowRequestMonitoringEnabled = slowRequestMonitoringEnabled;
+        this.slowRequestThresholdMs = slowRequestThresholdMs;
+        this.dumpStack = dumpStack;
+        this.timer = timer;
     }
 
     @Override
@@ -84,7 +99,11 @@ protected void initChannel(SocketChannel ch) throws Exception {
                                 endpointListenerName,
                                 isInternal,
                                 requestsMetrics,
-                                serverAuthenticator));
+                                serverAuthenticator,
+                                slowRequestMonitoringEnabled,
+                                slowRequestThresholdMs,
+                                dumpStack,
+                                timer));
     }
 
     @Override

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/SlowRequestDetector.java

@@ -23,6 +23,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -44,7 +46,7 @@ class SlowRequestDetector extends TimerTask {
     private static final Logger LOG = LoggerFactory.getLogger(SlowRequestDetector.class);
 
     private final FlussRequest request;
-    private final Thread processingThread;
+    private volatile Thread processingThread;
     private final boolean dumpStack;
     private final long thresholdMs;
     private final AtomicBoolean executed = new AtomicBoolean(false);
@@ -53,14 +55,15 @@ class SlowRequestDetector extends TimerTask {
      * Creates a slow request detector.
      *
      * @param request the request to monitor
-     * @param processingThread the thread processing this request
+     * @param processingThread the thread processing this request (can be null initially and set
+     *     later)
      * @param thresholdMs the threshold in milliseconds after which the request is considered slow
      * @param dumpStack whether to dump the thread stack trace
      * @param delayMs the delay in milliseconds before checking if the request is slow
      */
     SlowRequestDetector(
             FlussRequest request,
-            Thread processingThread,
+            @Nullable Thread processingThread,
             long thresholdMs,
             boolean dumpStack,
             long delayMs) {
@@ -71,6 +74,17 @@ class SlowRequestDetector extends TimerTask {
         this.dumpStack = dumpStack;
     }
 
+    /**
+     * Sets the processing thread. This allows the detector to be created when the request is
+     * received (to monitor queue time) and updated with the actual processing thread when the
+     * request starts being processed.
+     *
+     * @param processingThread the thread that is processing the request
+     */
+    void setProcessingThread(Thread processingThread) {
+        this.processingThread = processingThread;
+    }
+
     @Override
     public void run() {
         // Ensure this task only executes once even if there are race conditions