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 a78333f0698a CAMEL-26049: camel-coap - destroy the CoAP servers when
the component stops, and the client endpoint of a producer when it stops
(#28536)
a78333f0698a is described below
commit a78333f0698a4b45c2a7b677662191015af3aa5c
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 27 15:41:59 2026 +0530
CAMEL-26049: camel-coap - destroy the CoAP servers when the component
stops, and the client endpoint of a producer when it stops (#28536)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/camel/coap/CoAPComponent.java | 5 +-
.../java/org/apache/camel/coap/CoAPProducer.java | 36 ++++++--
.../coap/CoAPComponentStopDestroysServersTest.java | 102 +++++++++++++++++++++
.../camel/coap/CoAPProducerNoResponseTest.java | 61 ++++++++++++
.../coap/CoAPProducerStopDestroysEndpointTest.java | 63 +++++++++++++
5 files changed, 257 insertions(+), 10 deletions(-)
diff --git
a/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPComponent.java
b/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPComponent.java
index 19adfe6da90b..4a1d41cf588b 100644
---
a/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPComponent.java
+++
b/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPComponent.java
@@ -304,9 +304,12 @@ public class CoAPComponent extends DefaultComponent
implements RestConsumerFacto
@Override
protected void doStop() throws Exception {
+ // destroy (not only stop) the servers: stop() keeps their executors,
whose threads are not daemon threads
+ // and would keep the JVM running after the CamelContext is stopped; a
later start creates them again
for (CoapServer s : servers.values()) {
- s.stop();
+ s.destroy();
}
+ servers.clear();
super.doStop();
}
}
diff --git
a/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPProducer.java
b/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPProducer.java
index c563fceb47d3..e539607eaf44 100644
---
a/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPProducer.java
+++
b/components/camel-coap/src/main/java/org/apache/camel/coap/CoAPProducer.java
@@ -26,13 +26,16 @@ import org.apache.camel.util.URISupport;
import org.eclipse.californium.core.CoapClient;
import org.eclipse.californium.core.CoapResponse;
import org.eclipse.californium.core.coap.MediaTypeRegistry;
+import org.eclipse.californium.core.network.Endpoint;
/**
* The CoAP producer.
*/
public class CoAPProducer extends DefaultProducer {
private final CoAPEndpoint endpoint;
- private CoapClient client;
+ // set by initClient under the lock, and set to null by doStop, which
BaseService.stop runs under the same lock;
+ // volatile for the readers that do not take the lock (getClient)
+ private volatile CoapClient client;
private boolean shutdownClient;
public CoAPProducer(CoAPEndpoint endpoint) {
@@ -50,7 +53,16 @@ public class CoAPProducer extends DefaultProducer {
@Override
public void process(Exchange exchange) throws Exception {
- initClient();
+ // doStop sets the client field to null, possibly while this request
is in flight: take the client once, under
+ // the lock that doStop holds, so it cannot become null between
initClient and this read
+ CoapClient coapClient;
+ lock.lock();
+ try {
+ initClient();
+ coapClient = client;
+ } finally {
+ lock.unlock();
+ }
String ct = exchange.getIn().getHeader(CoAPConstants.CONTENT_TYPE,
String.class);
if (ct == null) {
@@ -58,27 +70,27 @@ public class CoAPProducer extends DefaultProducer {
ct = "application/octet-stream";
}
// the method header may be given in lower case
- String method = CoAPHelper.getDefaultMethod(exchange,
client).toUpperCase(Locale.ROOT);
+ String method = CoAPHelper.getDefaultMethod(exchange,
coapClient).toUpperCase(Locale.ROOT);
int mediaType = MediaTypeRegistry.parse(ct);
CoapResponse response = null;
boolean pingResponse = false;
switch (method) {
case CoAPConstants.METHOD_GET:
- response = client.get();
+ response = coapClient.get();
break;
case CoAPConstants.METHOD_DELETE:
- response = client.delete();
+ response = coapClient.delete();
break;
case CoAPConstants.METHOD_POST:
byte[] bodyPost = exchange.getIn().getBody(byte[].class);
- response = client.post(bodyPost, mediaType);
+ response = coapClient.post(bodyPost, mediaType);
break;
case CoAPConstants.METHOD_PUT:
byte[] bodyPut = exchange.getIn().getBody(byte[].class);
- response = client.put(bodyPut, mediaType);
+ response = coapClient.put(bodyPut, mediaType);
break;
case CoAPConstants.METHOD_PING:
- pingResponse = client.ping();
+ pingResponse = coapClient.ping();
break;
default:
throw new IllegalArgumentException("Unsupported CoAP method: "
+ method);
@@ -89,7 +101,7 @@ public class CoAPProducer extends DefaultProducer {
} else if (!method.equals(CoAPConstants.METHOD_PING)) {
// the client returns null when no response was received (timeout,
rejected or cancelled request)
throw new CamelExchangeException(
- "No response received from CoAP server for " + method + ":
" + URISupport.sanitizeUri(client.getURI()),
+ "No response received from CoAP server for " + method + ":
" + URISupport.sanitizeUri(coapClient.getURI()),
exchange);
}
@@ -114,7 +126,13 @@ public class CoAPProducer extends DefaultProducer {
@Override
protected void doStop() throws Exception {
if (shutdownClient && client != null) {
+ // the client was created by this producer with its own endpoint
(createCoapClient), which
+ // CoapClient.shutdown() does not release (bound socket and
connector threads)
+ Endpoint clientEndpoint = client.getEndpoint();
client.shutdown();
+ if (clientEndpoint != null) {
+ clientEndpoint.destroy();
+ }
client = null;
}
}
diff --git
a/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPComponentStopDestroysServersTest.java
b/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPComponentStopDestroysServersTest.java
new file mode 100644
index 000000000000..663001f914a2
--- /dev/null
+++
b/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPComponentStopDestroysServersTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.coap;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.eclipse.californium.core.CoapClient;
+import org.eclipse.californium.core.CoapResponse;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * Stopping the CamelContext must release the CoAP servers of the component:
CoapServer.stop() keeps the executors of
+ * the server, whose threads are not daemon threads, so a server that is only
stopped (not destroyed) keeps the JVM
+ * running after the CamelContext is stopped.
+ */
+class CoAPComponentStopDestroysServersTest {
+
+ @RegisterExtension
+ static AvailablePortFinder.Port port = CoAPTestPorts.findUdpPort();
+
+ @Test
+ void stoppingTheCamelContextReleasesTheServerThreads() throws Exception {
+ Set<Thread> before = Thread.getAllStackTraces().keySet();
+ CamelContext context = createContext();
+ try {
+ context.start();
+ assertEquals("Hi", get());
+ } finally {
+ context.stop();
+ }
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ List<String> left = Thread.getAllStackTraces().keySet().stream()
+ .filter(t -> !before.contains(t) && t.isAlive() &&
!t.isDaemon()
+ && t.getName().startsWith("CoapServer"))
+ .map(Thread::getName).sorted().toList();
+ assertEquals(List.of(), left, "non-daemon CoapServer threads still
alive after CamelContext.stop()");
+ });
+ }
+
+ @Test
+ void aCamelContextStartedAgainServesAgain() throws Exception {
+ // control: the servers destroyed by the stop are created again by the
consumers when the context starts again
+ CamelContext context = createContext();
+ try {
+ context.start();
+ assertEquals("Hi", get());
+ context.stop();
+ context.start();
+ assertEquals("Hi", get());
+ } finally {
+ context.stop();
+ }
+ }
+
+ private static CamelContext createContext() throws Exception {
+ CamelContext context = new DefaultCamelContext();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("coap://localhost:%d/hello",
port.getPort()).setBody(constant("Hi"));
+ }
+ });
+ return context;
+ }
+
+ private static String get() throws Exception {
+ CoapClient client = new CoapClient("coap://localhost:" +
port.getPort() + "/hello");
+ try {
+ CoapResponse response = client.get();
+ assertNotNull(response, "no response from the CoAP server");
+ return response.getResponseText();
+ } finally {
+ client.shutdown();
+ }
+ }
+}
diff --git
a/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerNoResponseTest.java
b/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerNoResponseTest.java
index 408e1058645f..666aa6aa4824 100644
---
a/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerNoResponseTest.java
+++
b/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerNoResponseTest.java
@@ -16,6 +16,14 @@
*/
package org.apache.camel.coap;
+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.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
@@ -23,11 +31,15 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.AvailablePortFinder;
import org.eclipse.californium.core.CoapClient;
+import org.eclipse.californium.core.coap.MessageObserverAdapter;
+import org.eclipse.californium.core.coap.Request;
+import
org.eclipse.californium.core.network.interceptors.MessageInterceptorAdapter;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -61,6 +73,55 @@ public class CoAPProducerNoResponseTest extends
CoAPTestSupport {
MockEndpoint.assertIsSatisfied(context);
}
+ @Test
+ void testStoppingTheProducerWithARequestInFlightFailsTheExchange() throws
Exception {
+ CoAPEndpoint endpoint = context.getEndpoint(
+ String.format("coap://localhost:%d/TestResource",
unusedPort.getPort()), CoAPEndpoint.class);
+ CoAPProducer producer = (CoAPProducer) endpoint.createProducer();
+ producer.start();
+ producer.initClient();
+
+ AtomicReference<Request> sentRequest = new AtomicReference<>();
+ CountDownLatch requestSent = new CountDownLatch(1);
+ producer.getClient().getEndpoint().addInterceptor(new
MessageInterceptorAdapter() {
+ @Override
+ public void sendRequest(Request request) {
+ // called before the connector sends the datagram
+ request.addMessageObserver(new MessageObserverAdapter() {
+ @Override
+ public void onSent(boolean retransmission) {
+ sentRequest.set(request);
+ requestSent.countDown();
+ }
+ });
+ }
+ });
+
+ Exchange exchange = endpoint.createExchange();
+ exchange.getIn().setHeader(CoAPConstants.COAP_METHOD,
CoAPConstants.METHOD_GET);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> inflight = executor.submit(() -> {
+ producer.process(exchange);
+ return null;
+ });
+ assertTrue(requestSent.await(5, TimeUnit.SECONDS), "the request
was not sent");
+
+ // destroying the client endpoint cancels the request; the
requesting thread waits for the response on the
+ // monitor of the request, so holding it here lets that thread go
on only after doStop has completed
+ Request request = sentRequest.get();
+ synchronized (request) {
+ producer.stop();
+ }
+
+ ExecutionException failure =
assertThrows(ExecutionException.class, () -> inflight.get(5, TimeUnit.SECONDS));
+ CamelExchangeException cause =
assertInstanceOf(CamelExchangeException.class, failure.getCause());
+ assertTrue(cause.getMessage().startsWith("No response received
from CoAP server for GET"), cause.getMessage());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
diff --git
a/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerStopDestroysEndpointTest.java
b/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerStopDestroysEndpointTest.java
new file mode 100644
index 000000000000..193133789dbd
--- /dev/null
+++
b/components/camel-coap/src/test/java/org/apache/camel/coap/CoAPProducerStopDestroysEndpointTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.coap;
+
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Each start of a CoAP producer creates a CoapClient with its own
CoapEndpoint (a bound UDP socket and its receiver and
+ * sender threads). CoapClient.shutdown() does not destroy that endpoint, so
the producer must do it when it stops.
+ */
+class CoAPProducerStopDestroysEndpointTest extends CoAPTestSupport {
+
+ @Test
+ void routeRestartsReleaseTheEndpointsOfTheStoppedProducers() throws
Exception {
+ assertEquals("Hello Camel", template.requestBody("direct:start",
"Camel", String.class));
+ Set<Thread> before = Thread.getAllStackTraces().keySet();
+ for (int i = 0; i < 5; i++) {
+ context.getRouteController().stopRoute("producer");
+ context.getRouteController().startRoute("producer");
+ assertEquals("Hello Camel", template.requestBody("direct:start",
"Camel", String.class));
+ }
+ context.getRouteController().stopRoute("producer");
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ long receivers = Thread.getAllStackTraces().keySet().stream()
+ .filter(t -> !before.contains(t) && t.isAlive() &&
t.getName().startsWith("UDP-Receiver"))
+ .count();
+ assertEquals(0, receivers, "UDP receiver threads of stopped CoAP
producers still alive after 5 restarts");
+ });
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("coap://localhost:%d/hello",
PORT.getPort()).transform(simple("Hello ${body}"));
+
from("direct:start").routeId("producer").toF("coap://localhost:%d/hello",
PORT.getPort());
+ }
+ };
+ }
+}
