This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 3750d4b83995 CAMEL-26574: camel-netty - Client mode consumer: close
the connection at stop, reconnect without blocking the event loop (#29199)
3750d4b83995 is described below
commit 3750d4b83995790d45dc7fe7d71cf6ec463b7f64
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 5 12:34:29 2026 +0530
CAMEL-26574: camel-netty - Client mode consumer: close the connection at
stop, reconnect without blocking the event loop (#29199)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../ClientModeTCPNettyServerBootstrapFactory.java | 177 +++++++++++--
.../netty/NettyConsumerClientModeStopTest.java | 284 +++++++++++++++++++++
2 files changed, 436 insertions(+), 25 deletions(-)
diff --git
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
index eee5bcbec220..ec7be44930b9 100644
---
a/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
+++
b/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
@@ -18,8 +18,13 @@ package org.apache.camel.component.netty;
import java.net.ConnectException;
import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
@@ -31,6 +36,7 @@ import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.ScheduledFuture;
import org.apache.camel.CamelContext;
import org.apache.camel.support.service.ServiceSupport;
import org.slf4j.Logger;
@@ -48,8 +54,14 @@ public class ClientModeTCPNettyServerBootstrapFactory
extends ServiceSupport imp
private NettyServerBootstrapConfiguration configuration;
private ChannelInitializer<Channel> pipelineFactory;
private Bootstrap clientBootstrap;
- private Channel channel;
private EventLoopGroup workerGroup;
+ // guards the fields below: the lifecycle thread and the event loops
(reconnect) use them
+ private final Lock reconnectLock = new ReentrantLock();
+ private Channel channel;
+ private ChannelFuture pendingConnect;
+ private ScheduledFuture<?> reconnectTask;
+ private boolean stopping;
+ private int generation;
public ClientModeTCPNettyServerBootstrapFactory() {
}
@@ -134,6 +146,14 @@ public class ClientModeTCPNettyServerBootstrapFactory
extends ServiceSupport imp
LOG.debug("Created ClientBootstrap {}", clientBootstrap);
clientBootstrap.handler(pipelineFactory);
+ reconnectLock.lock();
+ try {
+ // a new start: callbacks of an earlier start do nothing
+ stopping = false;
+ generation++;
+ } finally {
+ reconnectLock.unlock();
+ }
ChannelFuture channelFuture
= clientBootstrap.connect(new
InetSocketAddress(configuration.getHost(), configuration.getPort()));
@@ -141,25 +161,109 @@ public class ClientModeTCPNettyServerBootstrapFactory
extends ServiceSupport imp
configuration.getHost(), configuration.getPort(),
clientBootstrap);
LOG.info("ClientModeServerBootstrap binding to {}:{}",
configuration.getHost(), configuration.getPort());
- channel = openChannel(channelFuture);
+ Channel answer = openChannel(channelFuture);
+ reconnectLock.lock();
+ try {
+ if (answer != null && channel == null) {
+ channel = answer;
+ }
+ } finally {
+ reconnectLock.unlock();
+ }
}
protected void stopServerBootstrap() {
// close all channels
LOG.info("ClientModeServerBootstrap unbinding from {}:{}",
configuration.getHost(), configuration.getPort());
+ // the worker group may be shared (workerGroup option), so the
channel, a connect in progress and a
+ // scheduled reconnect must be closed and cancelled here and not left
to the shutdown of the group
+ List<Channel> channels = new ArrayList<>(2);
+ reconnectLock.lock();
+ try {
+ stopping = true;
+ if (reconnectTask != null) {
+ reconnectTask.cancel(false);
+ reconnectTask = null;
+ }
+ if (pendingConnect != null) {
+ channels.add(pendingConnect.channel());
+ pendingConnect = null;
+ }
+ if (channel != null) {
+ channels.add(channel);
+ channel = null;
+ }
+ } finally {
+ reconnectLock.unlock();
+ }
+ for (Channel ch : channels) {
+ try {
+ ChannelFuture closeFuture = ch.close();
+ // cannot wait on the event loop of the channel (a route
stopped from its own processing)
+ if (!ch.eventLoop().inEventLoop()) {
+
closeFuture.awaitUninterruptibly(configuration.getShutdownTimeout());
+ }
+ } catch (RuntimeException e) {
+ // a channel that never got an event loop (its creation or
registration failed) has nothing to close
+ LOG.debug("Cannot close channel {} due to {}. This exception
is ignored.", ch, e.getMessage(), e);
+ }
+ }
+
if (workerGroup != null) {
workerGroup.shutdownGracefully();
workerGroup = null;
}
}
+ /**
+ * Connects again if the channel is not active and the consumer is not
stopping. Does not wait for the connect (it
+ * may run on an event loop): the outcome is handled by a listener.
+ */
protected void doReconnectIfNeeded() throws Exception {
- if (channel == null || !channel.isActive()) {
+ doReconnectIfNeeded(currentGeneration());
+ }
+
+ private void doReconnectIfNeeded(int gen) {
+ // resolve the address before taking the lock (may do a DNS lookup)
+ InetSocketAddress address = new
InetSocketAddress(configuration.getHost(), configuration.getPort());
+ reconnectLock.lock();
+ try {
+ // a stopped or restarted consumer, or a connect already in
progress, does not connect again
+ if (stopping || gen != generation || pendingConnect != null ||
(channel != null && channel.isActive())) {
+ return;
+ }
LOG.debug("ClientModeServerBootstrap re-connect to {}:{}",
configuration.getHost(), configuration.getPort());
- ChannelFuture connectFuture
- = clientBootstrap.connect(new
InetSocketAddress(configuration.getHost(), configuration.getPort()));
- channel = openChannel(connectFuture);
+ final ChannelFuture connectFuture =
clientBootstrap.connect(address);
+ pendingConnect = connectFuture;
+ connectFuture.addListener((ChannelFutureListener) f ->
onReconnectComplete(f, gen));
+ } finally {
+ reconnectLock.unlock();
+ }
+ }
+
+ private void onReconnectComplete(ChannelFuture connectFuture, int gen) {
+ reconnectLock.lock();
+ try {
+ if (pendingConnect == connectFuture) {
+ pendingConnect = null;
+ }
+ if (stopping || gen != generation) {
+ // the consumer was stopped or restarted while connecting
+ connectFuture.channel().close();
+ return;
+ }
+ if (connectFuture.isSuccess()) {
+ channel = connectFuture.channel();
+ LOG.debug("ClientModeServerBootstrap re-connected to {}",
configuration.getAddress());
+ closeFutureReconnect(channel, gen);
+ } else {
+ LOG.debug("Re-connect to {} failed. Will attempt again in {}
millis.", configuration.getAddress(),
+ configuration.getReconnectInterval(),
connectFuture.cause());
+ scheduleReconnect(eventLoopOf(connectFuture.channel()), gen);
+ }
+ } finally {
+ reconnectLock.unlock();
}
}
@@ -172,10 +276,11 @@ public class ClientModeTCPNettyServerBootstrapFactory
extends ServiceSupport imp
// wait for the channel to be open (see io.netty.channel.ChannelFuture
javadoc for example/recommendation)
channelFuture.awaitUninterruptibly();
+ int gen = currentGeneration();
if (!channelFuture.isDone() || !channelFuture.isSuccess()) {
//check if reconnect is enabled and schedule a reconnect, if from
handler then don't schedule a reconnect
if (configuration.isReconnect()) {
- scheduleReconnect(channelFuture);
+ scheduleReconnect(eventLoopOf(channelFuture.channel()), gen);
return null;
} else {
ConnectException cause = new ConnectException("Cannot connect
to " + configuration.getAddress());
@@ -191,31 +296,53 @@ public class ClientModeTCPNettyServerBootstrapFactory
extends ServiceSupport imp
LOG.debug("Creating connector to address: {}",
configuration.getAddress());
}
+ closeFutureReconnect(answer, gen);
+ return answer;
+ }
+
+ private void closeFutureReconnect(Channel answer, int gen) {
// schedule a reconnect to happen when the channel closes
if (configuration.isReconnect()) {
- answer.closeFuture().addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws
Exception {
- scheduleReconnect(channelFuture);
+ answer.closeFuture().addListener((ChannelFutureListener) f ->
scheduleReconnect(answer.eventLoop(), gen));
+ }
+ }
+
+ private void scheduleReconnect(final EventLoop loop, final int gen) {
+ reconnectLock.lock();
+ try {
+ if (stopping || gen != generation) {
+ return;
+ }
+ reconnectTask = loop.schedule(() -> {
+ try {
+ LOG.trace("Re-connecting to {} if needed",
configuration.getAddress());
+ doReconnectIfNeeded(gen);
+ } catch (Exception e) {
+ LOG.warn("Error during re-connect to {}. Will attempt
again in {} millis. This exception is ignored.",
+ configuration.getAddress(),
configuration.getReconnectInterval(),
+ e);
+ scheduleReconnect(loop, gen);
}
- });
+ }, configuration.getReconnectInterval(), TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ LOG.debug("Cannot schedule a re-connect to {} as the event loop is
shutting down", configuration.getAddress());
+ } finally {
+ reconnectLock.unlock();
}
+ }
- return answer;
+ private EventLoop eventLoopOf(Channel ch) {
+ // a channel whose registration failed has no event loop
+ return ch.isRegistered() ? ch.eventLoop() :
clientBootstrap.config().group().next();
}
- private void scheduleReconnect(final ChannelFuture channelFuture) {
- final EventLoop loop = channelFuture.channel().eventLoop();
- loop.schedule(() -> {
- try {
- LOG.trace("Re-connecting to {} if needed",
configuration.getAddress());
- doReconnectIfNeeded();
- } catch (Exception e) {
- LOG.warn("Error during re-connect to {}. Will attempt again in
{} millis. This exception is ignored.",
- configuration.getAddress(),
configuration.getReconnectInterval(),
- e);
- }
- }, configuration.getReconnectInterval(), TimeUnit.MILLISECONDS);
+ private int currentGeneration() {
+ reconnectLock.lock();
+ try {
+ return generation;
+ } finally {
+ reconnectLock.unlock();
+ }
}
}
diff --git
a/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyConsumerClientModeStopTest.java
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyConsumerClientModeStopTest.java
new file mode 100644
index 000000000000..0e00533629f8
--- /dev/null
+++
b/components/camel-netty/src/test/java/org/apache/camel/component/netty/NettyConsumerClientModeStopTest.java
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.netty;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.string.StringEncoder;
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A consumer with clientMode=true closes its connection and stops
reconnecting when it stops, also with a shared worker
+ * group, and reconnects with a single worker thread.
+ */
+class NettyConsumerClientModeStopTest extends BaseNettyTest {
+
+ @BindToRegistry("sharedWorkerGroup")
+ private final EventLoopGroup sharedWorkerGroup
+ = new
NettyWorkerPoolBuilder().withWorkerCount(2).withName("SharedClientWorker").build();
+
+ @BindToRegistry("sharedSingleWorkerGroup")
+ private final EventLoopGroup sharedSingleWorkerGroup
+ = new
NettyWorkerPoolBuilder().withWorkerCount(1).withName("SharedSingleClientWorker").build();
+
+ // holds the event loop of the consumer once its reconnect has started to
connect
+ @BindToRegistry("holdReconnect")
+ private final ServerInitializerFactory holdReconnect = new
HoldReconnectInitializerFactory();
+ private final AtomicInteger initializedChannels = new AtomicInteger();
+ private final CountDownLatch reconnectStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseEventLoop = new CountDownLatch(1);
+
+ // the server side of each connection the consumer opened, in order
+ private final BlockingQueue<Channel> connections = new
LinkedBlockingQueue<>();
+ private EventLoopGroup serverBossGroup;
+ private EventLoopGroup serverWorkerGroup;
+
+ NettyConsumerClientModeStopTest() {
+ testConfiguration().withUseRouteBuilder(false);
+ }
+
+ @AfterEach
+ void shutdownGroups() {
+ releaseEventLoop.countDown();
+ // stop the consumers before the event loops they use
+ context.stop();
+ if (serverBossGroup != null) {
+ serverBossGroup.shutdownGracefully(0, 1,
TimeUnit.SECONDS).awaitUninterruptibly(10, TimeUnit.SECONDS);
+ serverWorkerGroup.shutdownGracefully(0, 1,
TimeUnit.SECONDS).awaitUninterruptibly(10, TimeUnit.SECONDS);
+ }
+ sharedWorkerGroup.shutdownGracefully(0, 1,
TimeUnit.SECONDS).awaitUninterruptibly(10, TimeUnit.SECONDS);
+ sharedSingleWorkerGroup.shutdownGracefully(0, 1,
TimeUnit.SECONDS).awaitUninterruptibly(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void stopClosesTheConnectionWithASharedWorkerGroup() throws Exception {
+ startServer();
+ addClientRoute("workerGroup=#sharedWorkerGroup&reconnect=false");
+ Channel first = nextConnection();
+
+ context.getRouteController().stopRoute("client");
+
+ assertTrue(first.closeFuture().await(10, TimeUnit.SECONDS), "the
connection of the stopped consumer is closed");
+ }
+
+ @Test
+ void noReconnectAfterStopWithASharedWorkerGroup() throws Exception {
+ startServer();
+
addClientRoute("workerGroup=#sharedWorkerGroup&reconnect=true&reconnectInterval=100");
+ Channel first = nextConnection();
+
+ context.getRouteController().stopRoute("client");
+ // the server drops the connection (if the stop did not close it)
+ first.close().sync();
+
+ // ten reconnect intervals
+ assertNull(connections.poll(1, TimeUnit.SECONDS), "a connection opened
after the stop");
+ }
+
+ @Test
+ void stopWhileReconnectingWithASharedWorkerGroup() throws Exception {
+ startServer();
+ addClientRoute(
+
"workerGroup=#sharedSingleWorkerGroup&reconnect=true&reconnectInterval=100&serverInitializerFactory=#holdReconnect");
+ nextConnection().close().sync();
+ // the reconnect has started to connect, and its event loop is held
before the connect completes
+ assertTrue(reconnectStarted.await(10, TimeUnit.SECONDS), "the
reconnect started");
+
+ context.getRouteController().stopRoute("client");
+ releaseEventLoop.countDown();
+
+ // a connection the reconnect made anyway is closed, and none follows
+ Channel late = connections.poll(1, TimeUnit.SECONDS);
+ if (late != null) {
+ assertTrue(late.closeFuture().await(10, TimeUnit.SECONDS),
+ "the connection of the reconnect in progress at the stop
is closed");
+ }
+ assertNull(connections.poll(1, TimeUnit.SECONDS), "a connection opened
after the stop");
+ }
+
+ @Test
+ void restartHasOneConnectionWithASharedWorkerGroup() throws Exception {
+ startServer();
+
addClientRoute("workerGroup=#sharedWorkerGroup&reconnect=true&reconnectInterval=100");
+ Channel first = nextConnection();
+
+ context.getRouteController().stopRoute("client");
+ context.getRouteController().startRoute("client");
+ Channel second = nextConnection();
+
+ assertTrue(first.closeFuture().await(10, TimeUnit.SECONDS), "the
connection of the first start is closed");
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("Hello");
+ second.writeAndFlush("Hello\n");
+ mock.assertIsSatisfied();
+ assertNull(connections.poll(1, TimeUnit.SECONDS), "a third
connection");
+ }
+
+ @Test
+ void reconnectsAfterTheServerDroppedTheConnectionWithASharedWorkerGroup()
throws Exception {
+ startServer();
+
addClientRoute("workerGroup=#sharedWorkerGroup&reconnect=true&reconnectInterval=100");
+ nextConnection().close().sync();
+
+ Channel second = nextConnection();
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("Hello");
+ second.writeAndFlush("Hello\n");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void twoConsumersReconnectWithASharedWorkerGroup() throws Exception {
+ startServer();
+ addClientRoute("client",
"workerGroup=#sharedWorkerGroup&reconnect=true&reconnectInterval=10");
+ addClientRoute("client2",
"workerGroup=#sharedWorkerGroup&reconnect=true&reconnectInterval=11");
+
+ // the server drops both connections at once, a few times: the
reconnects run at the same time on both event
+ // loops of the group, and each connects a channel that may be on the
other loop
+ Channel first = nextConnection();
+ Channel second = nextConnection();
+ for (int i = 0; i < 5; i++) {
+ first.close();
+ second.close().sync();
+ first = nextConnection();
+ second = nextConnection();
+ }
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceivedInAnyOrder("Hello", "World");
+ first.writeAndFlush("Hello\n");
+ second.writeAndFlush("World\n");
+ mock.assertIsSatisfied();
+ // one connection per consumer: no reconnect chain was duplicated
+ assertNull(connections.poll(1, TimeUnit.SECONDS), "a third
connection");
+ }
+
+ @Test
+ void reconnectsWithOneWorkerThread() throws Exception {
+ // the server is not up yet: the consumer starts and schedules a
reconnect
+ addClientRoute("workerCount=1&reconnect=true&reconnectInterval=100");
+ startServer();
+
+ Channel first = nextConnection();
+ MockEndpoint mock = getMockEndpoint("mock:received");
+ mock.expectedBodiesReceived("Hello", "World");
+ first.writeAndFlush("Hello\n");
+ // and again after the server dropped the connection
+ first.close().sync();
+ nextConnection().writeAndFlush("World\n");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void stopClosesTheConnectionWithItsOwnWorkerGroup() throws Exception {
+ startServer();
+ addClientRoute("reconnect=true&reconnectInterval=100");
+ Channel first = nextConnection();
+
+ context.getRouteController().stopRoute("client");
+
+ assertTrue(first.closeFuture().await(10, TimeUnit.SECONDS), "the
connection of the stopped consumer is closed");
+ assertNull(connections.poll(1, TimeUnit.SECONDS), "a connection opened
after the stop");
+ }
+
+ private final class HoldReconnectInitializerFactory extends
ServerInitializerFactory {
+
+ @Override
+ public ServerInitializerFactory createPipelineFactory(NettyConsumer
consumer) {
+ return new DefaultServerInitializerFactory(consumer) {
+ @Override
+ protected void initChannel(Channel ch) throws Exception {
+ super.initChannel(ch);
+ if (initializedChannels.incrementAndGet() == 2) {
+ // the channel of the first reconnect: hold its event
loop until the test releases it
+ ch.eventLoop().execute(() -> {
+ try {
+ releaseEventLoop.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ reconnectStarted.countDown();
+ }
+ }
+ };
+ }
+
+ @Override
+ protected void initChannel(Channel ch) {
+ throw new UnsupportedOperationException("only used to create the
pipeline factory of the consumer");
+ }
+ }
+
+ private Channel nextConnection() throws InterruptedException {
+ Channel channel = connections.poll(10, TimeUnit.SECONDS);
+ assertNotNull(channel, "a connection from the consumer");
+ return channel;
+ }
+
+ private void addClientRoute(String options) throws Exception {
+ addClientRoute("client", options);
+ }
+
+ private void addClientRoute(String routeId, String options) throws
Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("netty:tcp://localhost:{{port}}?textline=true&clientMode=true&" +
options).routeId(routeId)
+ .to("mock:received");
+ }
+ });
+ context.start();
+ }
+
+ private void startServer() throws InterruptedException {
+ serverBossGroup = new
NettyWorkerPoolBuilder().withWorkerCount(1).withName("TestServerBoss").build();
+ serverWorkerGroup = new
NettyWorkerPoolBuilder().withWorkerCount(1).withName("TestServerWorker").build();
+ new ServerBootstrap().group(serverBossGroup,
serverWorkerGroup).channel(NioServerSocketChannel.class)
+ .childHandler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ ch.pipeline().addLast(new StringEncoder());
+ ch.pipeline().addLast(new
ChannelInboundHandlerAdapter() {
+ @Override
+ public void channelActive(ChannelHandlerContext
ctx) {
+ connections.add(ctx.channel());
+ }
+ });
+ }
+ }).bind("localhost", getPort()).sync();
+ }
+}
