You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "allthingssecurity (via GitHub)" <gi...@apache.org> on 2026/10/16 00:04:30 UTC
[PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
allthingssecurity opened a new pull request, #27919:
URL: https://github.com/apache/camel/pull/27919
# Description
[CAMEL-25680](https://issues.apache.org/jira/browse/CAMEL-25680)
The `docker:events` and `docker:stats` consumers start a streaming command with `exec(callback)`. docker-java reads the stream on its own thread, and the stream belongs to the callback (`ResultCallbackTemplate.onStart(Closeable)` keeps it, `close()` closes it).
What went wrong on main:
1. `doStop()` of both consumers called `close()` on the command only. In docker-java 3.7.1 `AbstrDockerCmd.close()` does nothing, and the callback was not kept. So the stream stayed open after a stop, one connection per stop, and the old callback kept routing what it received. When the route was started again the new consumer opened a second stream, and every Docker event was routed twice (n + 1 times after n restarts).
2. `DockerEventsConsumer.EventsCallback` implemented only `onNext`. When the stream failed (daemon restart, lost connection) docker-java logged one ERROR and closed it; when the daemon ended the stream nothing was logged. The consumer stayed started and never received an event again until the route was restarted. The same when the daemon could not be reached when the route started.
Changes:
- Both consumers keep the callback of their stream and close it in `doStop()`. The callback closes a stream that only opens after the stop (`ResultCallbackTemplate.onStart` reopens a closed template), and a closed callback routes nothing.
- `DockerEventsConsumer` handles `onError` and `onComplete`: it subscribes again after 1 second, then with a delay that doubles up to 30 seconds while subscribing fails. These are the delays of `KubernetesHelper.watchAgain` (camel-kubernetes `KubernetesHelper` lines 99-105 on main), which the kubernetes consumers use to watch again. The delay runs on a single-thread scheduled executor of the consumer, created in `doStart()` and shut down in `doStop()`; a subscription that completes while the consumer stops is closed again.
- Logging: a stream that was open and fails is logged once as a WARN with the stack trace, a failed attempt to subscribe again (the daemon is still down) as a WARN line without stack trace, a clean end as a WARN line. `super.onError()` is not called, as it logs an ERROR with the stack trace for every attempt.
What users can notice (upgrade guide 4.24, new `camel-docker` section):
- After a daemon restart or a lost connection the events consumer receives events again. The events of the time without a stream are not received: the new stream starts with the events from then on (the first stream still uses `since`, as before).
- A route that is stopped and started again receives each event once.
- New WARN lines while the daemon cannot be reached (at most one every 30 seconds after the first minute).
Tests:
- New `DockerEventsConsumerStreamTest` (5 tests), mocked `DockerClient` that captures the callbacks; the test drives the callbacks itself; every wait bounded (10 seconds), no sleeps:
- stop and start the route, then one event on each stream that is still open: routed once, and the first stream closed;
- stop before the stream opens: the stream is closed when it opens;
- `onError` on an open stream: a new subscription; that one fails before it opens (connection refused), so another one follows (after 2 seconds), whose event is routed;
- `onComplete`: a new subscription;
- `docker:stats`: stopping the route closes the stream.
- Without the change (main's consumers, same tests) all five fail: `mock://result Received message count. Expected: <1> but was: <2>`, `the stream opened after the stop was not closed ==> expected: <1> but was: <0>`, `stopping the route did not close the stats stream ==> expected: <1> but was: <0>`, and the two resubscription tests time out (`ConditionTimeout ... was not fulfilled within 10 seconds`).
- With the change the camel-docker unit tests pass: 54 run, 0 failures, 4 skipped (integration tests that need a daemon).
What is not changed:
- The stats stream also ends when its container stops, and it is not opened again: subscribing again to a stopped container would fail every time, so that needs its own rule (follow-up).
- `withSince(String.valueOf(processInitialEvent()))` passes milliseconds where the Docker API expects seconds, so the `CamelDockerInitialRange` option probably never replays past events. Not changed here: I read it, but could not check it against a daemon (follow-up).
- No new option for the delays.
What was not run: the module's integration tests and any test against a real Docker daemon (restart, lost connection).
# Target
- [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
# Tracking
- [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it).
(The ticket is drafted and not filed yet; CAMEL-25680 is a placeholder until it is.)
# Apache Camel coding standards and style
- [x] I checked that each commit in the pull request has a meaningful subject line and body.
- [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes.
(I built and tested camel-docker, including the formatter and import-sort plugins. I did not run the full root build.)
# AI-assisted contributions
- [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used.
This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer.
_Claude Code on behalf of allthingssecurity_
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4302723830
##########
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();
+ }
+}
Review Comment:
Filed as CAMEL-25732: the test with a mocked `exec()` that throws and then succeeds, and the DEBUG log of the throwable from the other thread.
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4291166405
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -68,24 +91,125 @@ private long processInitialEvent() {
@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(), e);
Review Comment:
With the default `NettyDockerCmdExecFactory`, a daemon that is down makes `exec()` throw right away: the connect is synchronous and `onError` is never called. So every retry ends up here and logs the full stack trace, every 30 seconds while the daemon is down. Could this log one WARN line, like the `onError` path does, and keep the stack trace at DEBUG?
```suggestion
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);
```
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -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 ScheduledExecutorService executor;
Review Comment:
Nit: `executor` is read from the docker-java thread in `subscribeAgain()`. Could it be `volatile` like `eventsCmd` and `callback`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4302183298
##########
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();
+ }
+}
Review Comment:
Agreed by the author as a small follow-up (a mocked `exec()` that throws, then the next attempt succeeds). Resolving to merge; @allthingssecurity, please open a JIRA for the follow-up that covers this test and the DEBUG log of the other thread.
_Claude Code on behalf of Claus Ibsen_
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -96,5 +221,55 @@ public void onNext(Event event) {
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) {
Review Comment:
One WARN line without the stack trace is intended for a daemon that is down (every retry would otherwise print one); the DEBUG log of the throwable for parity goes into the same follow-up as the test in the other thread.
_Claude Code on behalf of Claus Ibsen_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #27919:
URL: https://github.com/apache/camel/pull/27919
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4291295742
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -68,24 +91,125 @@ private long processInitialEvent() {
@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(), e);
Review Comment:
Thanks, done in ebd724e5a110: a failed attempt to subscribe again now logs one WARN line with the message, and the stack trace at DEBUG, as you suggested. camel-docker tests: 54 run, 0 failures, 4 skipped.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -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 ScheduledExecutorService executor;
Review Comment:
Done in ebd724e5a110: `executor` is now `volatile`, like `eventsCmd` and `callback`.
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4292682842
##########
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();
+ }
+}
Review Comment:
Fair point: the synchronous `exec()` throw path in `subscribeAgain` is covered by reading, not by a test here. I can add a test for it (a mocked `exec()` that throws, then the next attempt succeeds) as a small follow-up, so this approved PR does not change again.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -96,5 +221,55 @@ public void onNext(Event event) {
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) {
Review Comment:
Intentional: the stack trace of a failed resubscribe is logged at DEBUG on the next line (as requested in review), so the WARN stays one line while the daemon is down; the stack trace of a stream that fails after it was open is still logged once at WARN.
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27919:
URL: https://github.com/apache/camel/pull/27919#issuecomment-6212152860
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-docker`
- `docs`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 8 of 706 tested, 26 compile-only — current: 8 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **8 affected modules** (current approach: 8).
**Skip-tests mode would test 8 modules** (2 direct + 7 downstream), **skip tests for 26** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (8)</summary>
- `camel-docker` ← components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java, components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java, components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerStreamTest.java
- `camel-jbang-mcp` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` ← downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-tui` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` ← downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-launcher-container` ← downstream of org.apache.camel:camel-launcher
- `camel-yaml-dsl-validator` ← downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← downstream of org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (26)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (35 modules, 5m 41s total)</summary>
**Total reactor time:** 5m 41s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 50.9s | SUCCESS |
| Camel :: JBang :: MCP | 42.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 35.8s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 23.8s | SUCCESS |
| Camel :: Docker | 21.4s | SUCCESS |
| Camel :: Component DSL | 20.7s | SUCCESS |
| Camel :: YAML DSL :: Validator | 19.7s | SUCCESS |
| Camel :: YAML DSL | 19.2s | SUCCESS |
| Camel :: Docs | 15.2s | SUCCESS |
| Camel :: Kamelet Main | 12.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 12.3s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 9.0s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 8.2s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 8.1s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 7.5s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 6.8s | SUCCESS |
| Camel :: All Components Sync point | 5.5s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.6s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 3.0s | SUCCESS |
| Camel :: Catalog :: Maven | 2.9s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 2.5s | SUCCESS |
| Camel :: Assembly | 1.7s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.2s | SUCCESS |
| Camel :: Coverage | 1.2s | SUCCESS |
| Camel :: Catalog :: Console | 1.2s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.1s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.8s | SUCCESS |
| Camel :: JBang :: Main | 0.8s | SUCCESS |
| Camel :: JBang :: Integration tests | 0.8s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.5s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: JBang :: Plugin :: TUI | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (50.9s)
- `Camel :: JBang :: MCP` (42.5s)
- `Camel :: JBang :: Plugin :: Validate` (35.8s)
- `Camel :: Catalog :: Camel Catalog` (23.8s)
- `Camel :: Docker` (21.4s)
- `Camel :: Component DSL` (20.7s)
- `Camel :: YAML DSL :: Validator` (19.7s)
- `Camel :: YAML DSL` (19.2s)
- `Camel :: Docs` (15.2s)
- `Camel :: Kamelet Main` (12.5s)
- `Camel :: JBang :: Plugin :: Kubernetes` (12.3s)
- `Camel :: YAML DSL :: Deserializers` (9.0s)
- `Camel :: Catalog :: Camel Route Parser` (8.2s)
- `Camel :: JBang :: Plugin :: Testing` (8.1s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (7.5s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (6.8s)
- `Camel :: All Components Sync point` (5.5s)
- `Camel :: YAML DSL :: Maven Plugins` (3.6s)
- `Camel :: Catalog :: Suggest (deprecated)` (3.0s)
- `Camel :: Catalog :: Maven` (2.9s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/38857137494)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4291355788
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -96,5 +221,55 @@ public void onNext(Event event) {
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) {
Review Comment:
🔍 **Logging asymmetry: throwable discarded when `open=false`**
When `open=false`, the `LOG.warn` omits the throwable entirely:
```java
LOG.warn("Failed to subscribe to the Docker events of {}, trying again in {} ms: {}",
endpoint, delay, throwable.getMessage()); // no throwable arg → no stack trace
```
Compare with the `open=true` branch (passes `throwable` as last arg → SLF4J logs the stack trace) and the `subscribeAgain` catch block which adds an explicit `LOG.debug(..., e)`. Add a `LOG.debug` after the warn for diagnostic parity:
```suggestion
} else {
LOG.warn("Failed to subscribe to the Docker events of {}, trying again in {} ms: {}", endpoint, delay,
throwable.getMessage());
LOG.debug("Failed to subscribe to the Docker events of {}", endpoint, throwable);
}
```
##########
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();
+ }
+}
Review Comment:
⚠️ **Missing test: synchronous `exec()` throw path in `subscribeAgain`**
None of the five tests exercise the `catch (Exception e)` path inside `subscribeAgain`'s lambda where `subscribe()` / `exec()` throws synchronously. All existing tests either add the callback cleanly or call `onError`/`onComplete` on the callback — none configure `exec()` to throw.
This is the Netty default transport's reconnect path: `exec()` throws `RuntimeException(ConnectException)` instead of invoking `onError`. Consider adding a test where `exec()` throws on the second invocation (after the first stream's `onError` triggers resubscription), and verify a third callback is created after the backoff delay.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27919:
URL: https://github.com/apache/camel/pull/27919#issuecomment-6211129908
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25680: camel-docker - events consumer closes its stream on stop and subscribes again when it ends [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4292683246
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -96,5 +221,55 @@ public void onNext(Event event) {
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) {
Review Comment:
Intentional, but you are right that the stack trace is then lost entirely. When the stream was open, its failure is logged once at WARN with the stack trace. When it never opened (the daemon is down), every retry would otherwise log a full stack trace, so that WARN is one line. For consistency with the `subscribeAgain` catch, it would be better to also log the throwable at DEBUG there. I can add that in the same small follow-up as the test above.
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org