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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
