Migrate to Netty 4.2 (kroxylicious#1467)

SamBarker · web-flow · commit 56f2e5f9f0fb · 2025-10-24T03:36:12.000Z
* Migrate to Netty 4.2

---------

Signed-off-by: Sam Barker &lt;sam@quadrocket.co.uk&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -13,6 +13,8 @@ Format `<github issue/pr number>: <short description>`.
 * [#2681](https://github.com/kroxylicious/kroxylicious/pull/2681): Create a maven archetype for filter development io.kroxylicious:kroxylicious-filter-archetype
 * [#2778](https://github.com/kroxylicious/kroxylicious/pull/2778): The proxy now allocates a sessionId to connect client and server channels for logging purposes. Allowing users to track activity between downstream and upstream channels.
 * [#143](https://github.com/kroxylicious/kroxylicious/issues/143): Add support for Netty metrics
+* [#2809](https://github.com/kroxylicious/kroxylicious/pull/2809): Add optional configuration parameter to control the number of worker threads used by Netty.
+* [#1467](https://github.com/kroxylicious/kroxylicious/pull/1467): Migrate to Netty 4.2
 
 ## 0.16.0
 
diff --git a/DEV_GUIDE.md b/DEV_GUIDE.md
@@ -380,6 +380,29 @@ You'll see an API response.  If the service_timeout change is effective, the soc
 will continue for 3 minutes.  If `socat` terminates after about 10 seconds, the workaround
 has been applied ineffectively.
 
+## Running integration tests with IO_Uring
+
+THe integration test suite enables IO_Uring un-conditionally which may trigger issues with memory limits. Certain platforms e.g. Fedora default to running with `RLIMIT_MEMLOCK` set.
+
+If you see test failures such as 
+```shell
+[ERROR] Errors:
+[ERROR]   MockServerTest.testClientCanSendAndReceiveRPCToMock:47 » IllegalState failed to create a child event loop
+```
+or 
+```shell
+java.lang.IllegalStateException: failed to create a child event loop
+...
+Caused by: java.lang.RuntimeException: failed to allocate memory for io_uring ring; try raising memlock limit (see getrlimit(RLIMIT_MEMLOCK, ...) or ulimit -l): Cannot allocate memory
+```
+
+Raise the `RLIMIT_MEMLOCK` (see https://lwn.net/Articles/876288/ for a discussion on the merits or otherwise of the default) by adding entries to `/etc/security/limits.conf` (see https://access.redhat.com/solutions/61334 for details on the file) the updates will take effect in the next login shell.
+example config entry:
+```text
+* hard memlock unlimited
+* soft memlock unlimited
+```
+
 ## Running system tests locally
 
 ### Prerequisites
diff --git a/kroxylicious-integration-test-support/pom.xml b/kroxylicious-integration-test-support/pom.xml
@@ -88,7 +88,7 @@
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
-            <artifactId>netty-codec</artifactId>
+            <artifactId>netty-codec-base</artifactId>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
@@ -102,6 +102,10 @@
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-classes-kqueue</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-classes-io_uring</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
diff --git a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/EventGroupConfig.java b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/client/EventGroupConfig.java
@@ -7,19 +7,25 @@
 package io.kroxylicious.test.client;
 
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.IoHandlerFactory;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.epoll.Epoll;
-import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollIoHandler;
 import io.netty.channel.epoll.EpollServerSocketChannel;
 import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.channel.kqueue.KQueue;
-import io.netty.channel.kqueue.KQueueEventLoopGroup;
+import io.netty.channel.kqueue.KQueueIoHandler;
 import io.netty.channel.kqueue.KQueueServerSocketChannel;
 import io.netty.channel.kqueue.KQueueSocketChannel;
-import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.channel.uring.IoUring;
+import io.netty.channel.uring.IoUringIoHandler;
+import io.netty.channel.uring.IoUringServerSocketChannel;
+import io.netty.channel.uring.IoUringSocketChannel;
 
 public record EventGroupConfig(
                                Class<? extends SocketChannel> clientChannelClass,
@@ -29,7 +35,11 @@ public static EventGroupConfig create() {
         final Class<? extends SocketChannel> clientChannelClass;
         final Class<? extends ServerSocketChannel> serverChannelClass;
 
-        if (Epoll.isAvailable()) {
+        if (IoUring.isAvailable()) {
+            clientChannelClass = IoUringSocketChannel.class;
+            serverChannelClass = IoUringServerSocketChannel.class;
+        }
+        else if (Epoll.isAvailable()) {
             clientChannelClass = EpollSocketChannel.class;
             serverChannelClass = EpollServerSocketChannel.class;
         }
@@ -44,23 +54,28 @@ else if (KQueue.isAvailable()) {
         return new EventGroupConfig(clientChannelClass, serverChannelClass);
     }
 
-    private static EventLoopGroup newGroup(int nThreads) {
-        if (Epoll.isAvailable()) {
-            return new EpollEventLoopGroup(nThreads);
+    private static EventLoopGroup newGroup() {
+        final IoHandlerFactory ioHandlerFactory;
+        if (IoUring.isAvailable()) {
+            ioHandlerFactory = IoUringIoHandler.newFactory();
+        }
+        else if (Epoll.isAvailable()) {
+            ioHandlerFactory = EpollIoHandler.newFactory();
         }
         else if (KQueue.isAvailable()) {
-            return new KQueueEventLoopGroup(nThreads);
+            ioHandlerFactory = KQueueIoHandler.newFactory();
         }
         else {
-            return new NioEventLoopGroup(nThreads);
+            ioHandlerFactory = NioIoHandler.newFactory();
         }
+        return new MultiThreadIoEventLoopGroup(1, ioHandlerFactory);
     }
 
     public EventLoopGroup newWorkerGroup() {
-        return newGroup(1);
+        return newGroup();
     }
 
     public EventLoopGroup newBossGroup() {
-        return newGroup(1);
+        return newGroup();
     }
 }
diff --git a/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/server/MockServer.java b/kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/server/MockServer.java
@@ -125,7 +125,7 @@ public int start(int port, ResponsePayload response, SslContext serverSslContext
         final EventGroupConfig eventGroupConfig = EventGroupConfig.create();
         bossGroup = eventGroupConfig.newBossGroup();
         workerGroup = eventGroupConfig.newWorkerGroup();
-        serverHandler = new MockHandler(response);
+        serverHandler = new MockHandler(null);
         ServerBootstrap b = new ServerBootstrap();
         b.group(bossGroup, workerGroup)
                 .channel(eventGroupConfig.serverChannelClass())
@@ -143,6 +143,10 @@ public void initChannel(SocketChannel ch) {
                     }
                 });
 
+        if (response != null) {
+            addMockResponseForApiKey(response);
+        }
+
         // Start the server.
         ChannelFuture f;
         try {
diff --git a/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/mock/MockServerTest.java b/kroxylicious-integration-test-support/src/test/java/io/kroxylicious/mock/MockServerTest.java
@@ -14,6 +14,9 @@
 
 import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.protocol.ApiMessage;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
@@ -31,6 +34,19 @@
 class MockServerTest {
     private static final Map<ApiAndVersion, ApiMessage> responseSamples = ApiMessageSampleGenerator.createResponseSamples();
     private static final Map<ApiAndVersion, ApiMessage> requestSamples = ApiMessageSampleGenerator.createRequestSamples();
+    private static MockServer mockServer;
+    private static KafkaClient kafkaClient;
+
+    @BeforeAll
+    static void beforeAll() {
+        mockServer = MockServer.startOnRandomPort(null);
+        kafkaClient = new KafkaClient("127.0.0.1", mockServer.port());
+    }
+
+    @AfterAll
+    static void afterAll() {
+        mockServer.close();
+    }
 
     public static Stream<ApiAndVersion> allSupportedApiVersions() {
         return DataClasses.getRequestClasses().keySet().stream().flatMap(apiKeys -> {
@@ -40,16 +56,20 @@ public static Stream<ApiAndVersion> allSupportedApiVersions() {
         });
     }
 
+    @BeforeEach
+    void setUp() {
+        mockServer.clear();
+
+    }
+
     @ParameterizedTest
     @MethodSource("allSupportedApiVersions")
     void testClientCanSendAndReceiveRPCToMock(ApiAndVersion apiKey) throws Exception {
         ResponsePayload mockResponse = getResponse(apiKey);
-        try (var mockServer = MockServer.startOnRandomPort(mockResponse);
-                var kafkaClient = new KafkaClient("127.0.0.1", mockServer.port())) {
-            CompletableFuture<Response> future = kafkaClient.get(getRequest(apiKey));
-            Response clientResponse = future.get(10, TimeUnit.SECONDS);
-            assertEquals(mockResponse, clientResponse.payload());
-        }
+        mockServer.addMockResponseForApiKey(mockResponse);
+        CompletableFuture<Response> future = kafkaClient.get(getRequest(apiKey));
+        Response clientResponse = future.get(10, TimeUnit.SECONDS);
+        assertEquals(mockResponse, clientResponse.payload());
     }
 
     private ResponsePayload getResponse(ApiAndVersion apiAndVersion) {
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/IOUringIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/IOUringIT.java
@@ -28,7 +28,7 @@
  * Integration test that tests Kroxylicious ability to utilise Linux io_uring.
  */
 @ExtendWith(KafkaClusterExtension.class)
-@EnabledIf(value = "io.netty.incubator.channel.uring.IOUring#isAvailable", disabledReason = "IOUring is not available")
+@EnabledIf(value = "io.netty.channel.uring.IoUring#isAvailable", disabledReason = "IOUring is not available")
 class IOUringIT extends BaseIT {
 
     private static final String HELLO_WORLD = "helloworld";
diff --git a/kroxylicious-runtime/pom.xml b/kroxylicious-runtime/pom.xml
@@ -95,7 +95,7 @@
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
-            <artifactId>netty-codec</artifactId>
+            <artifactId>netty-codec-base</artifactId>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
@@ -120,7 +120,7 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-native-epoll</artifactId>
-            <classifier>${netty.epoll.classifier}</classifier>
+            <classifier>${netty.epoll.architecture}</classifier>
             <scope>runtime</scope>
         </dependency>
         <dependency>
@@ -130,21 +130,21 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-native-kqueue</artifactId>
-            <classifier>${netty.kqueue.classifier}</classifier>
+            <classifier>${netty.kqueue.architecture}</classifier>
             <scope>runtime</scope>
         </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-transport-native-unix-common</artifactId>
         </dependency>
         <dependency>
-            <groupId>io.netty.incubator</groupId>
-            <artifactId>netty-incubator-transport-classes-io_uring</artifactId>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-classes-io_uring</artifactId>
         </dependency>
         <dependency>
-            <groupId>io.netty.incubator</groupId>
-            <artifactId>netty-incubator-transport-native-io_uring</artifactId>
-            <classifier>${netty.io_uring.classifier}</classifier>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport-native-io_uring</artifactId>
+            <classifier>${netty.io_uring.architecture}</classifier>
             <scope>runtime</scope>
         </dependency>
         <dependency>
@@ -186,6 +186,11 @@
             <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit-pioneer</groupId>
+            <artifactId>junit-pioneer</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
diff --git a/pom.xml b/pom.xml
