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 002db3d8b2d0 CAMEL-26048: camel-mina - dispose the acceptor of a
consumer when it stops, so a route restart does not leak it (#28535)
002db3d8b2d0 is described below
commit 002db3d8b2d0626fcd30e5cae89dd19222599d40
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 27 15:41:32 2026 +0530
CAMEL-26048: camel-mina - dispose the acceptor of a consumer when it stops,
so a route restart does not leak it (#28535)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/component/mina/MinaConsumer.java | 72 ++++++--
.../component/mina/MinaConsumerRestartTest.java | 205 +++++++++++++++++++++
2 files changed, 261 insertions(+), 16 deletions(-)
diff --git
a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java
b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java
index 2573fe8ec14c..cb9350edaea4 100644
---
a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java
+++
b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java
@@ -71,6 +71,8 @@ public class MinaConsumer extends DefaultConsumer {
private IoAcceptor acceptor;
private final MinaConfiguration configuration;
private ExecutorService workerPool;
+ // the acceptor or connector and the worker pool were released by doStop,
and are created again on start
+ private boolean released;
public MinaConsumer(final MinaEndpoint endpoint, Processor processor)
throws Exception {
super(endpoint, processor);
@@ -81,6 +83,10 @@ public class MinaConsumer extends DefaultConsumer {
//
this.getEndpoint().setExchangePattern(ExchangePattern.InOut);
+ setupProtocol();
+ }
+
+ private void setupProtocol() throws Exception {
String protocol = configuration.getProtocol();
if (protocol.equals("tcp")) {
if (configuration.isClientMode()) {
@@ -104,6 +110,10 @@ public class MinaConsumer extends DefaultConsumer {
@Override
protected void doStart() throws Exception {
super.doStart();
+ if (released) {
+ setupProtocol();
+ released = false;
+ }
if (configuration.isClientMode() &&
configuration.getProtocol().equals("tcp")) {
connector.setHandler(new ReceiveHandler());
ConnectFuture future = connector.connect(address);
@@ -120,26 +130,44 @@ public class MinaConsumer extends DefaultConsumer {
@Override
protected void doStop() throws Exception {
- if (configuration.isClientMode() &&
configuration.getProtocol().equals("tcp")) {
- LOG.info("Disconnect from server address: {} using connector: {}",
address, connector);
- if (session != null) {
- CloseFuture closeFuture = session.closeNow();
- closeFuture.awaitUninterruptibly();
- }
- connector.dispose(true);
- } else {
- LOG.info("Unbinding from server address: {} using acceptor: {}",
address, acceptor);
- if (address instanceof InetSocketAddress) {
- // need to check if the address is IPV4 all network address
- if ("0.0.0.0".equals(((InetSocketAddress)
address).getAddress().getHostAddress())) {
- LOG.info("Unbind the server address {}",
acceptor.getLocalAddresses());
- acceptor.unbind(acceptor.getLocalAddresses());
+ try {
+ if (configuration.isClientMode() &&
configuration.getProtocol().equals("tcp")) {
+ LOG.info("Disconnect from server address: {} using connector:
{}", address, connector);
+ if (session != null) {
+ CloseFuture closeFuture = session.closeNow();
+ closeFuture.awaitUninterruptibly();
+ }
+ } else {
+ LOG.info("Unbinding from server address: {} using acceptor:
{}", address, acceptor);
+ if (address instanceof InetSocketAddress) {
+ // need to check if the address is IPV4 all network address
+ if ("0.0.0.0".equals(((InetSocketAddress)
address).getAddress().getHostAddress())) {
+ LOG.info("Unbind the server address {}",
acceptor.getLocalAddresses());
+ acceptor.unbind(acceptor.getLocalAddresses());
+ } else {
+ acceptor.unbind(address);
+ }
} else {
acceptor.unbind(address);
}
- } else {
- acceptor.unbind(address);
}
+ } finally {
+ // a route restart creates a new consumer and never shuts this one
down (doShutdown), so release the
+ // connector or acceptor (selectors, I/O threads) and the worker
pool now, also when the disconnect or
+ // unbind failed; doStart creates them again if this consumer is
started again. The address is already
+ // unbound, so only wait for the I/O threads of the acceptor to
end when the CamelContext stops, as
+ // doShutdown did before (a datagram acceptor takes about two
seconds to end); once disposed, the dispose
+ // in doShutdown does nothing
+ if (connector != null) {
+ connector.dispose(true);
+ }
+ if (acceptor != null) {
+ acceptor.dispose(getEndpoint().getCamelContext().isStopping());
+ }
+ if (workerPool != null) {
+ workerPool.shutdown();
+ }
+ released = true;
}
super.doStop();
}
@@ -385,6 +413,18 @@ public class MinaConsumer extends DefaultConsumer {
return acceptor;
}
+ /**
+ * The connector of a consumer in client mode (for tests).
+ */
+ IoConnector getConnector() {
+ return connector;
+ }
+
+ /**
+ * Sets the acceptor this consumer binds when it starts. The consumer owns
it from then on: it is disposed when the
+ * consumer stops, and a start of the same consumer after a stop creates a
new acceptor from the configuration
+ * instead of binding this one again.
+ */
public void setAcceptor(IoAcceptor acceptor) {
this.acceptor = acceptor;
}
diff --git
a/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaConsumerRestartTest.java
b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaConsumerRestartTest.java
new file mode 100644
index 000000000000..d40d70ce0e6c
--- /dev/null
+++
b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaConsumerRestartTest.java
@@ -0,0 +1,205 @@
+/*
+ * 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.mina;
+
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.mina.core.service.IoAcceptor;
+import org.apache.mina.core.service.IoConnector;
+import org.apache.mina.core.service.IoService;
+import org.apache.mina.filter.executor.ExecutorFilter;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * A route stop/start creates a new MinaConsumer, and the stopped one is never
shut down: its acceptor (selectors, I/O
+ * threads) and worker pool must be released when it stops.
+ */
+class MinaConsumerRestartTest extends BaseMinaTest {
+
+ private final int udpPort = getNextPort();
+ private final int vmPort = getNextPort();
+
+ @Test
+ void routeRestartReleasesTheAcceptorOfTheStoppedTcpConsumer() throws
Exception {
+ for (IoAcceptor acceptor : restart("tcp", 3)) {
+ assertReleased(acceptor);
+ }
+ // the consumer of the restarted route works
+ sendAndExpect("mina:tcp://localhost:" + getPort() +
"?sync=false&textline=true");
+ }
+
+ @Test
+ void routeRestartReleasesTheAcceptorOfTheStoppedUdpConsumer() throws
Exception {
+ addUdpRoute();
+ for (IoAcceptor acceptor : restart("udp", 3)) {
+ assertReleased(acceptor);
+ }
+ sendAndExpect("mina:udp://localhost:" + udpPort + "?sync=false");
+ }
+
+ @Test
+ void routeRestartReleasesTheAcceptorOfTheStoppedVmConsumer() throws
Exception {
+ for (IoAcceptor acceptor : restart("vm", 3)) {
+ assertTrue(acceptor.isDisposed(), "the acceptor of a stopped
consumer must be disposed");
+ }
+ sendAndExpect("mina:vm://localhost:" + vmPort +
"?sync=false&textline=true");
+ }
+
+ @Test
+ void theSameConsumerCanBeStartedAgain() throws Exception {
+ // a consumer that is stopped and started again (not replaced) creates
a new acceptor
+ MinaConsumer consumer = (MinaConsumer)
context.getRoute("tcp").getConsumer();
+ IoAcceptor first = consumer.getAcceptor();
+
+ consumer.stop();
+ consumer.start();
+
+ assertTrue(first.isDisposed());
+ assertNotSame(first, consumer.getAcceptor());
+ assertFalse(consumer.getAcceptor().isDisposed());
+ sendAndExpect("mina:tcp://localhost:" + getPort() +
"?sync=false&textline=true");
+ }
+
+ @Test
+ void theSameClientModeConsumerCanBeStartedAgain() throws Exception {
+ // a consumer in client mode (connected to the tcp route) that is
stopped and started again creates a new
+ // connector and connects again; before, the start failed as the
stopped connector was disposed
+ context.getRouteController().startRoute("client");
+ MinaConsumer consumer = (MinaConsumer)
context.getRoute("client").getConsumer();
+ IoConnector first = consumer.getConnector();
+
+ consumer.stop();
+ consumer.start();
+
+ assertReleased(first);
+ assertNotSame(first, consumer.getConnector());
+ assertFalse(consumer.getConnector().isDisposed());
+ assertEquals(1, consumer.getConnector().getManagedSessionCount(), "the
restarted consumer is connected");
+ }
+
+ @Test
+ void theAcceptorIsReleasedWhenTheUnbindFails() throws Exception {
+ // a failing unbind must not skip the release: the acceptor and the
worker pool are released, and a start of the
+ // same consumer creates a new acceptor instead of binding the old one
again
+ MinaConsumer consumer = (MinaConsumer)
context.getRoute("tcp").getConsumer();
+ IoAcceptor first = consumer.getAcceptor();
+ IoAcceptor failing = mock(IoAcceptor.class, delegatesTo(first));
+ doThrow(new IllegalStateException("unbind
failed")).when(failing).unbind(any(SocketAddress.class));
+ consumer.setAcceptor(failing);
+
+ Exception e = assertThrows(IllegalStateException.class,
consumer::stop);
+ assertEquals("unbind failed", e.getMessage());
+
+ assertReleased(first);
+ consumer.start();
+ assertNotSame(failing, consumer.getAcceptor());
+ sendAndExpect("mina:tcp://localhost:" + getPort() +
"?sync=false&textline=true");
+ }
+
+ @Test
+ void onlyACamelContextStopWaitsForTheIoThreads() throws Exception {
+ // the address is already unbound, so a route stop does not wait for
the I/O threads of the acceptor (a datagram
+ // acceptor can take seconds to end); a CamelContext stop waits for
them, as doShutdown did before
+ addUdpRoute();
+ IoAcceptor routeStopped = watchAcceptor("udp");
+ context.getRouteController().stopRoute("udp");
+ verify(routeStopped).dispose(false);
+ verify(routeStopped, never()).dispose(true);
+
+ context.getRouteController().startRoute("udp");
+ IoAcceptor contextStopped = watchAcceptor("udp");
+ context.stop();
+ verify(contextStopped, never()).dispose(false);
+ // doShutdown disposes again, which does nothing then
+ verify(contextStopped, atLeastOnce()).dispose(true);
+ }
+
+ private void addUdpRoute() throws Exception {
+ // only in the udp tests: a datagram acceptor takes seconds to end,
and the CamelContext stop waits for it
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("mina:udp://localhost:%d?sync=false",
udpPort).routeId("udp").to("mock:result");
+ }
+ });
+ }
+
+ private IoAcceptor watchAcceptor(String routeId) {
+ // the acceptor of the consumer, behind a mock that passes every call
on to it
+ MinaConsumer consumer = (MinaConsumer)
context.getRoute(routeId).getConsumer();
+ IoAcceptor acceptor = mock(IoAcceptor.class,
delegatesTo(consumer.getAcceptor()));
+ consumer.setAcceptor(acceptor);
+ return acceptor;
+ }
+
+ private static void assertReleased(IoService service) {
+ assertTrue(service.isDisposed(), "the acceptor of a stopped consumer
must be disposed");
+ ExecutorFilter workers = (ExecutorFilter)
service.getFilterChain().get("threadPool");
+ assertTrue(((ExecutorService) workers.getExecutor()).isShutdown(),
+ "the worker pool of a stopped consumer must be shut down");
+ }
+
+ private List<IoAcceptor> restart(String routeId, int times) throws
Exception {
+ List<IoAcceptor> stopped = new ArrayList<>();
+ for (int i = 0; i < times; i++) {
+ stopped.add(((MinaConsumer)
context.getRoute(routeId).getConsumer()).getAcceptor());
+ context.getRouteController().stopRoute(routeId);
+ context.getRouteController().startRoute(routeId);
+ }
+ assertNotSame(stopped.get(times - 1), ((MinaConsumer)
context.getRoute(routeId).getConsumer()).getAcceptor(),
+ "a route restart creates a new consumer");
+ return stopped;
+ }
+
+ private void sendAndExpect(String uri) throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+ template.sendBody(uri, "hello");
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("mina:tcp://localhost:%d?sync=false&textline=true",
getPort()).routeId("tcp").to("mock:result");
+ fromF("mina:vm://localhost:%d?sync=false&textline=true",
vmPort).routeId("vm").to("mock:result");
+
fromF("mina:tcp://localhost:%d?sync=false&textline=true&clientMode=true",
getPort()).routeId("client")
+ .autoStartup(false).to("mock:client");
+ }
+ };
+ }
+}
