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 0c1d6b11798c CAMEL-25680: camel-docker - events consumer closes its
stream on stop and subscribes again when it ends (#27919)
0c1d6b11798c is described below
commit 0c1d6b11798c7126d75f2be515d7a05165e41915
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 16:06:18 2026 +0530
CAMEL-25680: camel-docker - events consumer closes its stream on stop and
subscribes again when it ends (#27919)
docker-java reads an events or stats stream on its own thread and the
stream belongs to the callback given to
exec(). The events and stats consumers closed only the command on stop, and
AbstrDockerCmd.close() does nothing
(docker-java 3.7.1), so the stream stayed open and its callback kept
routing what it received: a route that was
stopped and started again received every event once more for each restart,
and each stop left a connection open.
The events consumer had no onError / onComplete either: when the stream
failed or ended (Docker daemon restarted,
connection lost) it stayed started without receiving any event, with one
ERROR logged by docker-java for a failure and
nothing for a clean end.
Now both consumers close the stream of their callback on stop (also when
the stream only opens after the stop), and a
closed callback routes nothing. The events consumer subscribes again when
its stream fails or ends: after 1 second,
then with a delay that doubles up to 30 seconds while subscribing fails (as
the kubernetes consumers watch again). The
new stream starts with the events from then on; the events of the time
without a stream are not received.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../docker/consumer/DockerEventsConsumer.java | 185 ++++++++++++++++++++-
.../docker/consumer/DockerStatsConsumer.java | 42 ++++-
.../consumer/DockerEventsConsumerStreamTest.java | 182 ++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
4 files changed, 411 insertions(+), 6 deletions(-)
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
index 47596abdca02..62b329a5b502 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
@@ -16,6 +16,12 @@
*/
package org.apache.camel.component.docker.consumer;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
import com.github.dockerjava.api.async.ResultCallback;
import com.github.dockerjava.api.command.EventsCmd;
import com.github.dockerjava.api.model.Event;
@@ -36,9 +42,26 @@ public class DockerEventsConsumer extends DefaultConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(DockerEventsConsumer.class);
+ /**
+ * The delay before the consumer subscribes again after its event stream
failed or ended.
+ */
+ static final long SUBSCRIBE_AGAIN_DELAY_MILLIS = 1000;
+
+ /**
+ * The maximum delay between two attempts to subscribe again, when
subscribing keeps failing (for example while the
+ * Docker daemon cannot be reached).
+ */
+ static final long SUBSCRIBE_AGAIN_MAX_DELAY_MILLIS = 30000;
+
private DockerEndpoint endpoint;
private DockerComponent component;
- private EventsCmd eventsCmd;
+ private volatile EventsCmd eventsCmd;
+ /**
+ * The callback of the current event stream. docker-java opens the stream
on its own thread and the stream belongs
+ * to this callback: closing the command does not close it.
+ */
+ private volatile EventsCallback callback;
+ private volatile ScheduledExecutorService executor;
public DockerEventsConsumer(DockerEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -68,24 +91,126 @@ public class DockerEventsConsumer extends DefaultConsumer {
@Override
protected void doStart() throws Exception {
- this.eventsCmd = DockerClientFactory.getDockerClient(component,
endpoint.getConfiguration(), null).eventsCmd()
- .withSince(String.valueOf(processInitialEvent()));
- this.eventsCmd.exec(new EventsCallback());
+ executor =
getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "DockerEventsConsumer");
+ try {
+ subscribe(String.valueOf(processInitialEvent()),
SUBSCRIBE_AGAIN_DELAY_MILLIS);
+ } catch (Exception e) {
+ // a failed start does not call doStop()
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ executor = null;
+ throw e;
+ }
super.doStart();
}
@Override
protected void doStop() throws Exception {
- this.eventsCmd.close();
+ if (executor != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ executor = null;
+ }
+ EventsCallback current = callback;
+ callback = null;
+ if (current != null) {
+ current.close();
+ }
+ if (eventsCmd != null) {
+ eventsCmd.close();
+ }
super.doStop();
}
+ /**
+ * Opens a new event stream.
+ *
+ * @param since the value of the since parameter, or <tt>null</tt> to
receive the events from now on
+ * @param retryDelay the delay before subscribing again if this stream
fails before it is open
+ */
+ private void subscribe(String since, long retryDelay) {
+ EventsCmd cmd = DockerClientFactory.getDockerClient(component,
endpoint.getConfiguration(), null).eventsCmd();
+ if (since != null) {
+ cmd = cmd.withSince(since);
+ }
+ EventsCallback newCallback = new EventsCallback(retryDelay);
+ this.eventsCmd = cmd;
+ this.callback = newCallback;
+ cmd.exec(newCallback);
+ if (!isRunAllowed()) {
+ // the consumer was stopped while it subscribed again, so stopping
could not close this stream
+ callback = null;
+ newCallback.closeQuietly();
+ }
+ }
+
+ /**
+ * Subscribes again after the delay, as long as the consumer runs. The
events of the time without a stream are not
+ * received: the new stream starts with the events from then on.
+ */
+ private void subscribeAgain(long delayMillis) {
+ ScheduledExecutorService current = executor;
+ if (!isRunAllowed() || current == null) {
+ return;
+ }
+ try {
+ current.schedule(() -> {
+ if (!isRunAllowed()) {
+ return;
+ }
+ long nextDelay = Math.min(delayMillis * 2,
SUBSCRIBE_AGAIN_MAX_DELAY_MILLIS);
+ try {
+ subscribe(null, nextDelay);
+ } catch (Exception e) {
+ LOG.warn("Failed to subscribe again to the Docker events
of {}, trying again in {} ms: {}",
+ endpoint, nextDelay, e.getMessage());
+ LOG.debug("Failed to subscribe again to the Docker events
of {}", endpoint, e);
+ subscribeAgain(nextDelay);
+ }
+ }, delayMillis, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ LOG.debug("Cannot subscribe again to the Docker events of {} as
the consumer is stopping", endpoint, e);
+ }
+ }
+
protected class EventsCallback extends ResultCallback.Adapter<Event> {
+ /**
+ * The delay before subscribing again when this stream fails before it
is open. Once it is open, it is the
+ * initial delay again.
+ */
+ private final long retryDelay;
+ private volatile boolean open;
+ private volatile boolean closed;
+
+ protected EventsCallback() {
+ this(SUBSCRIBE_AGAIN_DELAY_MILLIS);
+ }
+
+ /**
+ * @param retryDelay the delay before subscribing again when this
stream fails before it is open
+ */
+ EventsCallback(long retryDelay) {
+ this.retryDelay = retryDelay;
+ }
+
+ @Override
+ public void onStart(Closeable stream) {
+ super.onStart(stream);
+ open = true;
+ if (closed) {
+ // closed (the consumer was stopped) while the stream was
being opened: close the stream now
+ closeQuietly();
+ }
+ }
+
@Override
public void onNext(Event event) {
+ if (closed) {
+ // an event of a stream the consumer no longer reads
+ return;
+ }
LOG.debug("Received Docker Event: {}", event);
final Exchange exchange = createExchange(true);
@@ -96,5 +221,55 @@ public class DockerEventsConsumer extends DefaultConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+
+ @Override
+ public void onError(Throwable throwable) {
+ if (closed) {
+ // closed by the consumer, which makes the read of the stream
fail
+ return;
+ }
+ // not super.onError(), which logs an ERROR with the stack trace
for every failed attempt
+ closeQuietly();
+ if (callback != this || !isRunAllowed()) {
+ return;
+ }
+ long delay = open ? SUBSCRIBE_AGAIN_DELAY_MILLIS : retryDelay;
+ if (open) {
+ LOG.warn("The Docker event stream of {} failed, subscribing
again in {} ms: {}", endpoint, delay,
+ throwable.getMessage(), throwable);
+ } else {
+ LOG.warn("Failed to subscribe to the Docker events of {},
trying again in {} ms: {}", endpoint, delay,
+ throwable.getMessage());
+ }
+ subscribeAgain(delay);
+ }
+
+ @Override
+ public void onComplete() {
+ if (closed) {
+ return;
+ }
+ closeQuietly();
+ if (callback != this || !isRunAllowed()) {
+ return;
+ }
+ long delay = open ? SUBSCRIBE_AGAIN_DELAY_MILLIS : retryDelay;
+ LOG.warn("The Docker event stream of {} ended, subscribing again
in {} ms", endpoint, delay);
+ subscribeAgain(delay);
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ super.close();
+ }
+
+ private void closeQuietly() {
+ try {
+ close();
+ } catch (IOException e) {
+ LOG.debug("Error closing the Docker event stream of {}",
endpoint, e);
+ }
+ }
}
}
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
index cb05b8ae48f4..58810a321643 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.docker.consumer;
+import java.io.Closeable;
+import java.io.IOException;
+
import com.github.dockerjava.api.async.ResultCallbackTemplate;
import com.github.dockerjava.api.command.StatsCmd;
import com.github.dockerjava.api.model.Statistics;
@@ -42,6 +45,11 @@ public class DockerStatsConsumer extends DefaultConsumer {
private DockerEndpoint endpoint;
private DockerComponent component;
private StatsCmd statsCmd;
+ /**
+ * The callback of the statistics stream. docker-java opens the stream on
its own thread and the stream belongs to
+ * this callback: closing the command does not close it.
+ */
+ private volatile StatsCallback callback;
public DockerStatsConsumer(DockerEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -60,13 +68,20 @@ public class DockerStatsConsumer extends DefaultConsumer {
String.class);
this.statsCmd = DockerClientFactory.getDockerClient(component,
endpoint.getConfiguration(), null).statsCmd(containerId);
- this.statsCmd.exec(new StatsCallback());
+ StatsCallback newCallback = new StatsCallback();
+ this.callback = newCallback;
+ this.statsCmd.exec(newCallback);
super.doStart();
}
@Override
protected void doStop() throws Exception {
+ StatsCallback current = callback;
+ callback = null;
+ if (current != null) {
+ current.close();
+ }
this.statsCmd.close();
super.doStop();
@@ -74,8 +89,27 @@ public class DockerStatsConsumer extends DefaultConsumer {
protected class StatsCallback extends
ResultCallbackTemplate<StatsCallback, Statistics> {
+ private volatile boolean closed;
+
+ @Override
+ public void onStart(Closeable stream) {
+ super.onStart(stream);
+ if (closed) {
+ // closed (the consumer was stopped) while the stream was
being opened: close the stream now
+ try {
+ close();
+ } catch (IOException e) {
+ LOGGER.debug("Error closing the Docker statistics stream",
e);
+ }
+ }
+ }
+
@Override
public void onNext(Statistics statistics) {
+ if (closed) {
+ // statistics of a stream the consumer no longer reads
+ return;
+ }
LOGGER.debug("Received Docker Statistics Event: {}", statistics);
final Exchange exchange = createExchange(true);
@@ -86,5 +120,11 @@ public class DockerStatsConsumer extends DefaultConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ super.close();
+ }
}
}
diff --git
a/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerStreamTest.java
b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerStreamTest.java
new file mode 100644
index 000000000000..6a8691e34935
--- /dev/null
+++
b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerStreamTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.docker.consumer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.async.ResultCallback;
+import com.github.dockerjava.api.command.EventsCmd;
+import com.github.dockerjava.api.command.StatsCmd;
+import com.github.dockerjava.api.model.Event;
+import com.github.dockerjava.api.model.Statistics;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.docker.DockerComponent;
+import org.apache.camel.component.docker.DockerConfiguration;
+import org.apache.camel.component.docker.util.DockerTestUtils;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The events and stats consumers read a stream that docker-java opens on its
own thread. The stream belongs to the
+ * callback given to {@code exec}: closing the command does not close it.
+ */
+class DockerEventsConsumerStreamTest extends CamelTestSupport {
+
+ private final DockerClient dockerClient = mock(DockerClient.class);
+ private final List<ResultCallback<Event>> eventCallbacks = new
CopyOnWriteArrayList<>();
+ private final List<ResultCallback<Statistics>> statsCallbacks = new
CopyOnWriteArrayList<>();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ DockerConfiguration configuration = new DockerConfiguration();
+
configuration.setParameters(DockerTestUtils.getDefaultParameters("localhost",
2375, configuration));
+ DockerComponent component = new DockerComponent(configuration);
+ component.setClient(DockerTestUtils.getClientProfile("localhost",
2375, configuration), dockerClient);
+ camelContext.addComponent("docker", component);
+
+ when(dockerClient.eventsCmd()).thenAnswer(i -> {
+ EventsCmd cmd = mock(EventsCmd.class);
+ when(cmd.withSince(anyString())).thenReturn(cmd);
+ when(cmd.exec(any())).thenAnswer(e -> {
+ ResultCallback<Event> callback = e.getArgument(0);
+ eventCallbacks.add(callback);
+ return callback;
+ });
+ return cmd;
+ });
+ when(dockerClient.statsCmd(anyString())).thenAnswer(i -> {
+ StatsCmd cmd = mock(StatsCmd.class);
+ when(cmd.exec(any())).thenAnswer(e -> {
+ ResultCallback<Statistics> callback = e.getArgument(0);
+ statsCallbacks.add(callback);
+ return callback;
+ });
+ return cmd;
+ });
+ return camelContext;
+ }
+
+ @Test
+ void stopClosesTheEventStreamAndARestartReceivesEachEventOnce() throws
Exception {
+ addRoute("docker://events?host=localhost&port=2375");
+ AtomicInteger closed = new AtomicInteger();
+ eventCallbacks.get(0).onStart(closed::incrementAndGet);
+
+ context.getRouteController().stopRoute("docker");
+ context.getRouteController().startRoute("docker");
+ assertEquals(2, eventCallbacks.size());
+ eventCallbacks.get(1).onStart(() -> {
+ });
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ Event event = new Event("start", "c1", "image", 1L);
+ // the daemon sends the event to every stream that is still open
+ for (ResultCallback<Event> callback : eventCallbacks) {
+ callback.onNext(event);
+ }
+ mock.assertIsSatisfied();
+ assertEquals(1, closed.get(), "stopping the route did not close the
event stream");
+ }
+
+ @Test
+ void stopBeforeTheStreamIsOpenClosesItWhenItOpens() throws Exception {
+ addRoute("docker://events?host=localhost&port=2375");
+ context.getRouteController().stopRoute("docker");
+
+ AtomicInteger closed = new AtomicInteger();
+ eventCallbacks.get(0).onStart(closed::incrementAndGet);
+ assertEquals(1, closed.get(), "the stream opened after the stop was
not closed");
+ }
+
+ @Test
+ void eventStreamThatFailsIsOpenedAgain() throws Exception {
+ addRoute("docker://events?host=localhost&port=2375");
+ eventCallbacks.get(0).onStart(() -> {
+ });
+
+ // the connection to the daemon is lost (for example the daemon
restarts)
+ eventCallbacks.get(0).onError(new IOException("Connection reset by
peer"));
+ await().atMost(10, TimeUnit.SECONDS).until(() -> eventCallbacks.size()
== 2);
+
+ // the daemon is not back yet: the new stream fails before it is open,
and is tried again later
+ eventCallbacks.get(1).onError(new ConnectException("Connection
refused"));
+ await().atMost(10, TimeUnit.SECONDS).until(() -> eventCallbacks.size()
== 3);
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived(new Event("start", "c2", "image", 2L));
+ eventCallbacks.get(2).onStart(() -> {
+ });
+ eventCallbacks.get(2).onNext(new Event("start", "c2", "image", 2L));
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void eventStreamThatEndsIsOpenedAgain() throws Exception {
+ addRoute("docker://events?host=localhost&port=2375");
+ eventCallbacks.get(0).onStart(() -> {
+ });
+
+ // the daemon ends the stream (for example when it shuts down)
+ eventCallbacks.get(0).onComplete();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> eventCallbacks.size()
== 2);
+ }
+
+ @Test
+ void stopClosesTheStatsStream() throws Exception {
+
addRoute("docker://stats?host=localhost&port=2375&containerId=470b9b823e6c");
+ AtomicInteger closed = new AtomicInteger();
+ Closeable stream = closed::incrementAndGet;
+ statsCallbacks.get(0).onStart(stream);
+
+ context.getRouteController().stopRoute("docker");
+ assertEquals(1, closed.get(), "stopping the route did not close the
stats stream");
+ }
+
+ private void addRoute(String uri) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri).routeId("docker").to("mock:result");
+ }
+ });
+ context.start();
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 890e3d79c5fb..8077be87d3b3 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1013,6 +1013,14 @@ nothing was stored and the exchange completed without an
error. The requests are
the answer like the other state operations: the states are stored, and a
request that Dapr rejects (or a sidecar that
cannot be reached) fails the exchange.
+=== camel-docker - the events consumer subscribes again when its event stream
ends
+
+The `docker:events` consumer subscribes again when its event stream fails or
ends (for example when the Docker daemon
+restarts): after 1 second, then with a delay that doubles up to 30 seconds
while the daemon cannot be reached. Before,
+it stayed started without receiving any event. The events of the time without
a stream are not received. Stopping a
+`docker:events` or `docker:stats` consumer now closes its stream: before, the
stream stayed open, and a route that was
+stopped and started again received every event once more for each restart.
+
=== camel-influxdb2 - a write is synchronous and a rejected write fails the
exchange
The producer wrote with the batching write API of the InfluxDB client: the
point was put into a buffer of the client
