allthingssecurity opened a new pull request, #28386: URL: https://github.com/apache/camel/pull/28386
# Description [CAMEL-25872](https://issues.apache.org/jira/browse/CAMEL-25872) `KafkaShareConsumer` does not implement `Suspendable`, so a suspend of the CamelContext (`CamelContext.suspend()`, or the JMX `suspend` operation) stops the share consumer and the resume starts it again. When the consumer starts, it hands its fetch tasks to the component instead of submitting them if the CamelContext is not started (CAMEL-23023), and `AbstractKafkaComponent` submits them from `onCamelContextFullyStarted`. During a resume the CamelContext is in the Starting state, and a resume does not call the startup listeners again. So the tasks were never submitted: the route reported Started, no share consumer was created, and no record was consumed until the route was restarted. Nothing was logged. When the consumer health check is enabled, it reports the route DOWN. Change: `AbstractKafkaComponent` remembers that it was notified that the CamelContext started. The flag is reset when the component stops. A new `startConsumer(Runnable)` runs the task at once when the flag is set or the CamelContext is started, or keeps it pending until the notification. The CamelContext is still checked because a component added with `addComponent` to a started CamelContext is not notified that it started (only a component that the CamelContext starts itself is registered as a startup listener), so with the flag alone its consumers would never start. The share consumer and the Kafka consumer call it instead of checking whether the CamelContext is started, so a consumer that starts before the CamelContext is fully started still waits, as before. The Kafka consumer is `Suspendable` (it is paused, not stopped), so it was not affected. It uses the same method so that both consumers start the same way. `pendingConsumer` is unchanged. No option changes, so there are n o generated changes. Tests: - New `KafkaShareConsumerSuspendResumeTest` uses the `RecordingShareConsumer` (a `MockShareConsumer`) of `KafkaShareConsumerTest`. - `consumesAgainAfterCamelContextIsResumed` suspends and resumes the CamelContext, then adds a record. Without the change it fails: the route is Started, but no share consumer is created after the resume (`ConditionTimeout` after 10 seconds). With the change the record is routed. - `consumesWithAComponentAddedAfterCamelContextIsStarted` adds a `KafkaShareComponent` with `addComponent` to the started CamelContext, then a route that uses it. It passes on main (which checks `isStarted()`); it is there for the `isStarted()` check of `startConsumer`: with the flag alone the consumer is never started (`ConditionTimeout` after 10 seconds). - With the change the unit tests pass: camel-kafka-common 11, camel-kafka 225 (1 skipped), camel-kafka-share 29, with 0 failures and 0 errors. - A stop and start of the CamelContext still delays the consumer until the CamelContext is fully started, and it starts after that (checked in the test log). What was not run: the ITs (they need Docker); a test against a real Kafka broker with share groups; the full root build. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] 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): [CAMEL-25872](https://issues.apache.org/jira/browse/CAMEL-25872). # 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-kafka-common, camel-kafka and camel-kafka-share, including the formatter and import-sort plugins. No option metadata changed. 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]
