CVE-2026-49361

@@ -855,6 +855,16 @@ public class ConfigOptions {
                     .withDescription(
                             "The number of queued requests allowed for worker threads, before blocking the I/O threads.");
 
+    public static final ConfigOption<MemorySize> NETTY_SERVER_MAX_REQUEST_SIZE =
+            key("netty.server.max-request-size")
+                    .memoryType()
+                    .defaultValue(MemorySize.parse("100mb"))
+                    .withDescription(
+                            "The maximum size of a single request that the server can receive. "
+                                    + "This limits the maximum frame length at the Netty pipeline level "
+                                    + "to protect the server from malicious clients sending oversized requests "
+                                    + "that could exhaust server memory.");
+
     public static final ConfigOption<Duration> NETTY_CONNECTION_MAX_IDLE_TIME =
             key("netty.connection.max-idle-time")
                     .durationType()

@@ -29,22 +29,24 @@
  * by the server to handle the Kafka requests for the client.
  */
 public class KafkaChannelInitializer extends NettyChannelInitializer {
-    public static final int MAX_FRAME_LENGTH = 100 * 1024 * 1024; // 100MB
 
     private final RequestChannel[] requestChannels;
+    private final int maxRequestSize;
     private final LengthFieldPrepender prepender = new LengthFieldPrepender(4);
 
-    public KafkaChannelInitializer(RequestChannel[] requestChannels, long maxIdleTimeSeconds) {
+    public KafkaChannelInitializer(
+            RequestChannel[] requestChannels, long maxIdleTimeSeconds, int maxRequestSize) {
         super(maxIdleTimeSeconds);
         this.requestChannels = requestChannels;
+        this.maxRequestSize = maxRequestSize;
     }
 
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
         super.initChannel(ch);
         addIdleStateHandler(ch);
         ch.pipeline().addLast(prepender);
-        addFrameDecoder(ch, MAX_FRAME_LENGTH, 4);
+        addFrameDecoder(ch, maxRequestSize, 4);
         ch.pipeline().addLast("flowController", new FlowControlHandler());
         ch.pipeline().addLast(new KafkaCommandDecoder(requestChannels));
     }

@@ -53,7 +53,8 @@ public ChannelHandler createChannelHandler(
             RequestChannel[] requestChannels, String listenerName) {
         return new KafkaChannelInitializer(
                 requestChannels,
-                conf.get(ConfigOptions.KAFKA_CONNECTION_MAX_IDLE_TIME).getSeconds());
+                conf.get(ConfigOptions.KAFKA_CONNECTION_MAX_IDLE_TIME).getSeconds(),
+                (int) conf.get(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE).getBytes());
     }
 
     @Override

@@ -70,6 +70,7 @@ public ChannelHandler createChannelHandler(
                 listenerName.equals(conf.get(ConfigOptions.INTERNAL_LISTENER_NAME)),
                 requestsMetrics,
                 conf.get(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME).getSeconds(),
+                (int) conf.get(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE).getBytes(),
                 Optional.ofNullable(
                                 AuthenticationFactory.loadServerAuthenticatorSuppliers(conf)
                                         .get(listenerName))

@@ -42,6 +42,7 @@ final class ServerChannelInitializer extends NettyChannelInitializer {
     private final String endpointListenerName;
     private final boolean isInternal;
     private final RequestsMetrics requestsMetrics;
+    private final int maxRequestSize;
     private final Supplier<ServerAuthenticator> authenticatorSupplier;
 
     public ServerChannelInitializer(
@@ -51,21 +52,23 @@ public ServerChannelInitializer(
             boolean isInternal,
             RequestsMetrics requestsMetrics,
             long maxIdleTimeSeconds,
+            int maxRequestSize,
             Supplier<ServerAuthenticator> authenticatorSupplier) {
         super(maxIdleTimeSeconds);
         this.requestChannels = requestChannels;
         this.apiManager = apiManager;
         this.endpointListenerName = endpointListenerName;
         this.isInternal = isInternal;
         this.requestsMetrics = requestsMetrics;
+        this.maxRequestSize = maxRequestSize;
         this.authenticatorSupplier = authenticatorSupplier;
     }
 
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
         super.initChannel(ch);
         // initialBytesToStrip=0 to include the frame size field after decoding
-        addFrameDecoder(ch, Integer.MAX_VALUE, 0);
+        addFrameDecoder(ch, maxRequestSize, 0);
         addIdleStateHandler(ch);
         ServerAuthenticator serverAuthenticator = authenticatorSupplier.get();
         LOG.debug(

@@ -18,17 +18,29 @@
 package org.apache.fluss.rpc.netty.server;
 
 import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.cluster.ServerNode;
+import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
 import org.apache.fluss.metrics.groups.MetricGroup;
 import org.apache.fluss.metrics.util.NOPMetricsGroup;
 import org.apache.fluss.rpc.RpcServer;
 import org.apache.fluss.rpc.TestingGatewayService;
+import org.apache.fluss.rpc.messages.ApiVersionsRequest;
+import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
+import org.apache.fluss.rpc.netty.client.NettyClient;
+import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.utils.NetUtils;
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
 import java.util.List;
 
+import static org.apache.fluss.utils.NetUtils.getAvailablePort;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link NettyServer}. */
 public class NettyServerTest {
@@ -50,4 +62,55 @@ void testPortAsZero() throws Exception {
             assertThat(bindEndpoints.get(0).getPort()).isGreaterThan(0);
         }
     }
+
+    @Test
+    void testOversizedRequestRejected() throws Exception {
+        // Configure an extremely small max request size (16 bytes) so that even the
+        // initial API_VERSIONS handshake request exceeds the limit and is rejected
+        // by the server's LengthFieldBasedFrameDecoder.
+        Configuration conf = new Configuration();
+        conf.set(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE, MemorySize.parse("16b"));
+        conf.setInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 3);
+
+        MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+        try (NetUtils.Port availablePort = getAvailablePort()) {
+            ServerNode serverNode =
+                    new ServerNode(1, "localhost", availablePort.getPort(), ServerType.COORDINATOR);
+            try (NettyServer nettyServer =
+                            new NettyServer(
+                                    conf,
+                                    Collections.singleton(
+                                            new Endpoint(
+                                                    serverNode.host(),
+                                                    serverNode.port(),
+                                                    "INTERNAL")),
+                                    new TestingGatewayService(),
+                                    metricGroup,
+                                    RequestsMetrics.createCoordinatorServerRequestMetrics(
+                                            metricGroup));
+                    NettyClient nettyClient =
+                            new NettyClient(
+                                    new Configuration(),
+                                    TestingClientMetricGroup.newInstance(),
+                                    false)) {
+                nettyServer.start();
+
+                // The NettyClient will try to send an API_VERSIONS handshake request.
+                // Since the request size exceeds the server's 16-byte limit, the server's
+                // frame decoder will reject the frame and close the connection, causing
+                // the client request to fail.
+                ApiVersionsRequest request =
+                        new ApiVersionsRequest()
+                                .setClientSoftwareName("test")
+                                .setClientSoftwareVersion("1.0");
+                assertThatThrownBy(
+                                () ->
+                                        nettyClient
+                                                .sendRequest(
+                                                        serverNode, ApiKeys.API_VERSIONS, request)
+                                                .get())
+                        .hasMessageContaining("Adjusted frame length exceeds 16");
+            }
+        }
+    }
 }

@@ -93,13 +93,14 @@ during the Fluss cluster working.
 
 ## Netty
 
-| Option                           | Type     | Default | Description                                                                                                                                 |
-|----------------------------------|----------|---------|---------------------------------------------------------------------------------------------------------------------------------------------|
-| netty.server.num-network-threads | Integer  | 3       | The number of threads that the server uses for receiving requests from the network and sending responses to the network.                    |
-| netty.server.num-worker-threads  | Integer  | 8       | The number of threads that the server uses for processing requests, which may include disk and remote I/O.                                  |
-| netty.server.max-queued-requests | Integer  | 500     | The number of queued requests allowed for worker threads, before blocking the I/O threads.                                                  |
-| netty.connection.max-idle-time   | Duration | 10min   | Close idle connections after the given time specified by this config.                                                                       |
-| netty.client.num-network-threads | Integer  | 4       | The number of threads that the client uses for sending requests to the network and receiving responses from network. The default value is 4 |
+| Option                           | Type       | Default | Description                                                                                                                                                                                                   |
+|----------------------------------|------------|---------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| netty.server.num-network-threads | Integer    | 3       | The number of threads that the server uses for receiving requests from the network and sending responses to the network.                                                                                      |
+| netty.server.num-worker-threads  | Integer    | 8       | The number of threads that the server uses for processing requests, which may include disk and remote I/O.                                                                                                    |
+| netty.server.max-queued-requests | Integer    | 500     | The number of queued requests allowed for worker threads, before blocking the I/O threads.                                                                                                                    |
+| netty.server.max-request-size    | MemorySize | 100mb   | The maximum size of a single request that the server can receive. This limits the maximum frame length at the Netty pipeline level to protect the server from malicious clients sending oversized requests that could exhaust server memory. |
+| netty.connection.max-idle-time   | Duration   | 10min   | Close idle connections after the given time specified by this config.                                                                                                                                         |
+| netty.client.num-network-threads | Integer    | 4       | The number of threads that the client uses for sending requests to the network and receiving responses from network. The default value is 4.                                                                  |
 
 ## Log
 

@@ -11,4 +11,31 @@ If you have concerns regarding Fluss's security or discover a vulnerability or p
 
 In the email, specify the project name **Fluss** and include a description of the issue or potential threat. You are also encouraged to include steps to reproduce the issue. The security team and the Fluss community will get back to you after assessing and analyzing the findings.
 
-**PLEASE PAY ATTENTION** to report the security issue privately to **security@apache.org** before disclosing it publicly.
\ No newline at end of file
+**PLEASE PAY ATTENTION** to report the security issue privately to **security@apache.org** before disclosing it publicly.
+
+## Security Updates
+
+This section lists fixed vulnerabilities in Fluss.
+
+<table class="table table-bordered">
+  <thead>
+    <tr>
+      <th class="text-left" width="200">CVE ID</th>
+      <th class="text-left" width="250">Affected Fluss versions</th>
+      <th class="text-left" width="550">Notes</th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr>
+      <td>
+        <a href="https://www.cve.org/CVERecord?id=CVE-2026-49361">CVE-2026-49361</a>
+      </td>
+      <td>
+        0.8.0, 0.9.0
+      </td>
+      <td>
+        Users are advised to upgrade to Fluss 0.9.1 or later versions. See the <a href="https://lists.apache.org/thread/dccw6tj0njwtmvbftq13mw7fdhsok373">advisory</a> for details.
+      </td>
+    </tr>
+  </tbody>
+</table>
\ No newline at end of file