You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/16 05:11:12 UTC
(paimon) branch master updated: [service] Fix connection race leak in NetworkClient.sendRequest (#8684)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new dfc2cd7919 [service] Fix connection race leak in NetworkClient.sendRequest (#8684)
dfc2cd7919 is described below
commit dfc2cd791937a8f614a62db0fceb5e121edcfb32
Author: Eunbin Son <58...@users.noreply.github.com>
AuthorDate: Thu Jul 16 14:11:07 2026 +0900
[service] Fix connection race leak in NetworkClient.sendRequest (#8684)
---
.../paimon/service/network/NetworkClient.java | 38 ++++---
.../paimon/service/network/NetworkClientTest.java | 121 +++++++++++++++++++++
2 files changed, 143 insertions(+), 16 deletions(-)
diff --git a/paimon-service/paimon-service-client/src/main/java/org/apache/paimon/service/network/NetworkClient.java b/paimon-service/paimon-service-client/src/main/java/org/apache/paimon/service/network/NetworkClient.java
index 748d1d2f05..a14971aa0a 100644
--- a/paimon-service/paimon-service-client/src/main/java/org/apache/paimon/service/network/NetworkClient.java
+++ b/paimon-service/paimon-service-client/src/main/java/org/apache/paimon/service/network/NetworkClient.java
@@ -143,22 +143,28 @@ public class NetworkClient<REQ extends MessageBody, RESP extends MessageBody> {
new IllegalStateException(clientName + " is already shut down."));
}
- ServerConnection<REQ, RESP> serverConnection = connections.get(serverAddress);
- if (serverConnection == null) {
- final ServerConnection<REQ, RESP> newConnection =
- ServerConnection.createPendingConnection(clientName, messageSerializer, stats);
- serverConnection = newConnection;
- connections.put(serverAddress, newConnection);
- bootstrap
- .connect(serverAddress.getAddress(), serverAddress.getPort())
- .addListener((ChannelFutureListener) newConnection::establishConnection);
-
- newConnection
- .getCloseFuture()
- .handle(
- (ignoredA, ignoredB) ->
- connections.remove(serverAddress, newConnection));
- }
+ final ServerConnection<REQ, RESP> serverConnection =
+ connections.computeIfAbsent(
+ serverAddress,
+ ignored -> {
+ final ServerConnection<REQ, RESP> newConnection =
+ ServerConnection.createPendingConnection(
+ clientName, messageSerializer, stats);
+ bootstrap
+ .connect(serverAddress.getAddress(), serverAddress.getPort())
+ .addListener(
+ (ChannelFutureListener)
+ newConnection::establishConnection);
+
+ newConnection
+ .getCloseFuture()
+ .handle(
+ (ignoredA, ignoredB) ->
+ connections.remove(
+ serverAddress, newConnection));
+
+ return newConnection;
+ });
return serverConnection.sendRequest(request);
}
diff --git a/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/network/NetworkClientTest.java b/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/network/NetworkClientTest.java
index 39e070fbee..efc08781bc 100644
--- a/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/network/NetworkClientTest.java
+++ b/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/network/NetworkClientTest.java
@@ -55,11 +55,13 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
@@ -332,6 +334,88 @@ class NetworkClientTest {
}
}
+ /**
+ * Tests that concurrent first requests racing on the initial connect to the same server address
+ * create exactly one connection. The atomic {@code computeIfAbsent} guarantees a single {@link
+ * ServerConnection}, hence a single accepted server channel and no channel leak after shutdown.
+ */
+ @Test
+ void testConcurrentFirstRequestsCreateSingleConnection() throws Exception {
+ AtomicServiceRequestStats stats = new AtomicServiceRequestStats();
+
+ final MessageSerializer<KvRequest, KvResponse> serializer =
+ new MessageSerializer<>(
+ new KvRequest.KvRequestDeserializer(),
+ new KvResponse.KvResponseDeserializer());
+
+ final KvResponse expected = KvResponseTest.random();
+
+ ExecutorService executor = null;
+ NetworkClient<KvRequest, KvResponse> client = null;
+ Channel serverChannel = null;
+
+ try {
+ final int numThreads = 8;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ client = new NetworkClient<>("Test Client", 1, serializer, stats);
+
+ final AtomicInteger acceptedChannels = new AtomicInteger(0);
+ serverChannel =
+ createServerChannel(
+ new CountingRespondingChannelHandler(
+ serializer, expected, acceptedChannels));
+
+ final InetSocketAddress serverAddress = getServerAddress(serverChannel);
+
+ // Release all threads simultaneously to maximize contention on the initial connect.
+ final CountDownLatch startLatch = new CountDownLatch(1);
+ final NetworkClient<KvRequest, KvResponse> finalClient = client;
+ Callable<CompletableFuture<KvResponse>> queryTask =
+ () -> {
+ startLatch.await();
+ KvRequest request = KvRequestTest.random();
+ return finalClient.sendRequest(serverAddress, request);
+ };
+
+ List<Future<CompletableFuture<KvResponse>>> submitted = new ArrayList<>();
+ for (int i = 0; i < numThreads; i++) {
+ submitted.add(executor.submit(queryTask));
+ }
+
+ startLatch.countDown();
+
+ // Wait for all request futures to complete successfully.
+ for (Future<CompletableFuture<KvResponse>> future : submitted) {
+ KvResponse actual = future.get().get();
+ assertThat(actual).isEqualTo(expected);
+ }
+
+ // computeIfAbsent guarantees a single ServerConnection, hence a single accepted
+ // channel.
+ assertThat(acceptedChannels.get()).isEqualTo(1);
+ } finally {
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ if (serverChannel != null) {
+ serverChannel.close();
+ }
+
+ if (client != null) {
+ try {
+ client.shutdown().get();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ assertThat(client.isEventGroupShutdown()).isTrue();
+ }
+
+ assertThat(stats.getNumConnections()).withFailMessage("Channel leak").isZero();
+ }
+ }
+
/**
* Tests that a server failure closes the connection and removes it from the established
* connections.
@@ -572,4 +656,41 @@ class NetworkClientTest {
ctx.channel().writeAndFlush(serResponse);
}
}
+
+ @ChannelHandler.Sharable
+ private static final class CountingRespondingChannelHandler
+ extends ChannelInboundHandlerAdapter {
+ private final MessageSerializer<KvRequest, KvResponse> serializer;
+ private final KvResponse response;
+ private final AtomicInteger acceptedChannels;
+
+ private CountingRespondingChannelHandler(
+ MessageSerializer<KvRequest, KvResponse> serializer,
+ KvResponse response,
+ AtomicInteger acceptedChannels) {
+ this.serializer = serializer;
+ this.response = response;
+ this.acceptedChannels = acceptedChannels;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) {
+ acceptedChannels.incrementAndGet();
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ ByteBuf buf = (ByteBuf) msg;
+ assertThat(MessageSerializer.deserializeHeader(buf)).isEqualTo(MessageType.REQUEST);
+ long requestId = MessageSerializer.getRequestId(buf);
+ serializer.deserializeRequest(buf);
+
+ buf.release();
+
+ ByteBuf serResponse =
+ MessageSerializer.serializeResponse(ctx.alloc(), requestId, response);
+
+ ctx.channel().writeAndFlush(serResponse);
+ }
+ }
}