This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new bf6f03959bb [BP-2.4][FLINK-41247][queryable-state] Close server
channels before shutting down the Netty event loops (#29699)
bf6f03959bb is described below
commit bf6f03959bb640c3a1388e2d73f9e7441b972fe0
Author: Purushottam Sinha <[email protected]>
AuthorDate: Mon Nov 2 20:26:52 2026 +0530
[BP-2.4][FLINK-41247][queryable-state] Close server channels before
shutting down the Netty event loops (#29699)
With a quiet period of 0, Netty 4.2 can terminate an event loop without
closing its registered channels if the shutdown starts while the loop runs
tasks (https://github.com/netty/netty/issues/17897). AbstractServerBase shut
its event loop group down this way, which leaked the listening socket and
accepted connections, and made AbstractServerTest.testPortRangeSuccess
flaky.
The server now tracks its server channel and accepted channels in a
ChannelGroup and closes them before shutting down the event loop group.
Generated-by: Claude Code (Claude Opus 5.5)
(cherry picked from commit d8af3821bd9f62717a87c22aa84c146ad4bacfe3)
---
.../queryablestate/network/AbstractServerBase.java | 49 ++++++++++++++----
.../queryablestate/network/AbstractServerTest.java | 58 ++++++++++++++++++++++
2 files changed, 98 insertions(+), 9 deletions(-)
diff --git
a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
index 1ebf06708cd..b6c1f85ceda 100644
---
a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
+++
b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
@@ -32,11 +32,14 @@ import
org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption;
import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
import
org.apache.flink.shaded.netty4.io.netty.channel.MultiThreadIoEventLoopGroup;
+import org.apache.flink.shaded.netty4.io.netty.channel.group.ChannelGroup;
+import
org.apache.flink.shaded.netty4.io.netty.channel.group.DefaultChannelGroup;
import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioIoHandler;
import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel;
import
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel;
import
org.apache.flink.shaded.netty4.io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import
org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler;
+import
org.apache.flink.shaded.netty4.io.netty.util.concurrent.ImmediateEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -93,6 +96,9 @@ public abstract class AbstractServerBase<REQ extends
MessageBody, RESP extends M
/** Netty's ServerBootstrap. */
private ServerBootstrap bootstrap;
+ /** The server channel and all accepted channels, closed before the event
loops shut down. */
+ private ChannelGroup channels;
+
/** Query executor thread pool. */
private ExecutorService queryExecutor;
@@ -241,6 +247,9 @@ public abstract class AbstractServerBase<REQ extends
MessageBody, RESP extends M
new MultiThreadIoEventLoopGroup(
numEventLoopThreads, threadFactory,
NioIoHandler.newFactory());
+ // stayClosed: a channel accepted concurrently with the shutdown is
closed once it's added
+ this.channels = new DefaultChannelGroup(serverName,
ImmediateEventExecutor.INSTANCE, true);
+
this.bootstrap =
new ServerBootstrap()
.localAddress(bindAddress, port)
@@ -248,7 +257,7 @@ public abstract class AbstractServerBase<REQ extends
MessageBody, RESP extends M
.channel(NioServerSocketChannel.class)
.option(ChannelOption.ALLOCATOR, bufferPool)
.childOption(ChannelOption.ALLOCATOR, bufferPool)
- .childHandler(new ServerChannelInitializer<>(handler));
+ .childHandler(new ServerChannelInitializer<>(handler,
channels));
final int defaultHighWaterMark = 64 * 1024; // from
DefaultChannelConfig (not exposed)
//noinspection ConstantConditions
@@ -264,6 +273,7 @@ public abstract class AbstractServerBase<REQ extends
MessageBody, RESP extends M
try {
final ChannelFuture future = bootstrap.bind().sync();
if (future.isSuccess()) {
+ channels.add(future.channel());
final InetSocketAddress localAddress =
(InetSocketAddress) future.channel().localAddress();
serverAddress =
@@ -314,15 +324,29 @@ public abstract class AbstractServerBase<REQ extends
MessageBody, RESP extends M
if (bootstrap != null) {
EventLoopGroup group = bootstrap.config().group();
if (group != null && !group.isShutdown()) {
- group.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS)
+ // Close the channels before shutting down the event
loops. With a quiet period
+ // of 0, Netty 4.2 can terminate an event loop without
closing its channels,
+ // which leaks their sockets
(https://github.com/netty/netty/issues/17897).
+ channels.close()
.addListener(
- finished -> {
- if (finished.isSuccess()) {
- groupShutdownFuture.complete(null);
- } else {
-
groupShutdownFuture.completeExceptionally(
- finished.cause());
+ closed -> {
+ if (!closed.isSuccess()) {
+ log.warn(
+ "Failed to close the
channels of {}.",
+ serverName,
+ closed.cause());
}
+ group.shutdownGracefully(0L, 0L,
TimeUnit.MILLISECONDS)
+ .addListener(
+ finished -> {
+ if
(finished.isSuccess()) {
+
groupShutdownFuture.complete(null);
+ } else {
+
groupShutdownFuture
+
.completeExceptionally(
+
finished.cause());
+ }
+ });
});
} else {
groupShutdownFuture.complete(null);
@@ -381,18 +405,25 @@ public abstract class AbstractServerBase<REQ extends
MessageBody, RESP extends M
/** The shared request handler. */
private final AbstractServerHandler<REQ, RESP> sharedRequestHandler;
+ /** The group that tracks the accepted channels. */
+ private final ChannelGroup channels;
+
/**
* Creates the channel pipeline initializer with the shared request
handler.
*
* @param sharedRequestHandler Shared request handler.
+ * @param channels The group that tracks the accepted channels.
*/
- ServerChannelInitializer(AbstractServerHandler<REQ, RESP>
sharedRequestHandler) {
+ ServerChannelInitializer(
+ AbstractServerHandler<REQ, RESP> sharedRequestHandler,
ChannelGroup channels) {
this.sharedRequestHandler =
Preconditions.checkNotNull(sharedRequestHandler,
"MessageBody handler");
+ this.channels = Preconditions.checkNotNull(channels);
}
@Override
protected void initChannel(SocketChannel channel) throws Exception {
+ channels.add(channel);
channel.pipeline()
.addLast(new ChunkedWriteHandler())
.addLast(new
LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4))
diff --git
a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
index edc60d185ac..165d9d3936a 100644
---
a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
+++
b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.queryablestate.network;
import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.queryablestate.network.messages.MessageBody;
import org.apache.flink.queryablestate.network.messages.MessageDeserializer;
import org.apache.flink.queryablestate.network.messages.MessageSerializer;
@@ -28,10 +29,13 @@ import
org.apache.flink.queryablestate.network.stats.KvStateRequestStats;
import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
+import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
import org.junit.jupiter.api.Test;
import java.net.InetAddress;
+import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Collections;
@@ -139,6 +143,47 @@ class AbstractServerTest {
assertThat(clientStats.getNumRequests()).isEqualTo(2L);
}
+ /**
+ * Tests that shutting down the server closes the server channel and the
accepted channels, also
+ * when the event loop is busy at the time of the shutdown.
+ */
+ @Test
+ void testShutdownClosesChannelsWhileEventLoopIsBusy() throws Throwable {
+ AtomicKvStateRequestStats serverStats = new
AtomicKvStateRequestStats();
+
+ try (TestServer server =
+ new TestServer(
+ "Test Server",
+ serverStats,
+ Collections.singletonList(0).iterator());
+ Socket client = new Socket()) {
+ server.start();
+ client.connect(server.getServerAddress());
+
+ Channel acceptedChannel = server.getAcceptedChannel().get();
+ assertThat(serverStats.getNumConnections()).isEqualTo(1L);
+
+ OneShotLatch taskRunning = new OneShotLatch();
+ OneShotLatch releaseTask = new OneShotLatch();
+ acceptedChannel
+ .eventLoop()
+ .execute(
+ () -> {
+ taskRunning.trigger();
+ releaseTask.awaitQuietly();
+ });
+ taskRunning.await();
+
+ CompletableFuture<Void> shutdownFuture = server.shutdownServer();
+ releaseTask.trigger();
+ shutdownFuture.get();
+
+ assertThat(acceptedChannel.isOpen()).isFalse();
+ assertThat(acceptedChannel.parent().isOpen()).isFalse();
+ assertThat(serverStats.getNumConnections()).isEqualTo(0L);
+ }
+ }
+
private static class TestClient extends Client<TestMessage, TestMessage>
implements AutoCloseable {
@@ -166,6 +211,8 @@ class AbstractServerTest {
private final KvStateRequestStats requestStats;
+ private final CompletableFuture<Channel> acceptedChannel = new
CompletableFuture<>();
+
TestServer(String name, KvStateRequestStats stats, Iterator<Integer>
bindPort)
throws UnknownHostException {
super(name, InetAddress.getLocalHost().getHostName(), bindPort, 1,
1);
@@ -181,6 +228,12 @@ class AbstractServerTest {
new TestMessage.TestMessageDeserializer()),
requestStats) {
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws
Exception {
+ super.channelActive(ctx);
+ acceptedChannel.complete(ctx.channel());
+ }
+
@Override
public CompletableFuture<TestMessage> handleRequest(
long requestId, TestMessage request) {
@@ -196,6 +249,11 @@ class AbstractServerTest {
};
}
+ /** Returns the first channel that this server accepted. */
+ CompletableFuture<Channel> getAcceptedChannel() {
+ return acceptedChannel;
+ }
+
@Override
public void close() throws Exception {
shutdownServer().get();
