[
https://issues.apache.org/jira/browse/CAMEL-25872?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131396#comment-18131396
]
Claus Ibsen commented on CAMEL-25872:
-------------------------------------
Merged to main in https://github.com/apache/camel/pull/28386 (commit
20c94435789e), fixed in 4.24.0.
_Claude Code on behalf of davsclaus_
> camel-kafka-share - after the CamelContext is suspended and resumed, the
> share consumer never consumes again (the route stays Started)
> --------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25872
> URL: https://issues.apache.org/jira/browse/CAMEL-25872
> Project: Camel
> Issue Type: Bug
> Components: camel-kafka
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> {{KafkaShareConsumer}} (main a0382c2a75ed) does not implement
> {{Suspendable}}, so a suspend of the CamelContext
> ({{CamelContext.suspend()}}, the JMX {{suspend}} operation of the
> CamelContext) stops the share consumer, and the resume starts it again
> ({{AbstractCamelContext.doResume}}, a route that does not support suspension
> is started instead of resumed).
> When the consumer starts, it hands its fetch tasks to the component instead
> of submitting them when the CamelContext is not started (CAMEL-23023, so that
> no record is processed before the CamelContext is fully started):
> {code:java}
> if (!endpoint.getCamelContext().isStarted()) {
> endpoint.getComponent().pendingConsumer(() -> executor.submit(task));
> } else {
> executor.submit(task);
> }
> {code}
> {{AbstractKafkaComponent}} submits the pending tasks from
> {{onCamelContextFullyStarted}}. During a resume the CamelContext is not in
> the Started state ({{BaseService.resume}} sets {{STARTING}} until
> {{doResume}} returns), so the tasks are added to the pending list, and a
> resume does not call the startup listeners again. The tasks are never
> submitted: the route reports Started, no share consumer is created and no
> record is consumed until the route or the CamelContext is restarted. Nothing
> is logged; the consumer health check (when enabled) reports the route DOWN,
> as its tasks are not connected.
> The Kafka consumer of camel-kafka implements {{Suspendable}} and is paused
> and resumed instead, so it is not affected.
> *Fix*: {{AbstractKafkaComponent}} remembers that it was notified that the
> CamelContext started (reset when the component stops), and a new
> {{startConsumer(Runnable)}} runs the task at once in that case or when 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 registered as a startup listener, so it is
> never notified. The share consumer and the Kafka consumer use it instead of
> checking whether the CamelContext is started, so the startup of a consumer
> before the CamelContext is fully started is delayed as before.
> *Tests*: new {{KafkaShareConsumerSuspendResumeTest}} (the
> {{RecordingShareConsumer}} of {{KafkaShareConsumerTest}}, a
> {{MockShareConsumer}}): suspend and resume the CamelContext, then add a
> record. On main it fails: the route is Started, but no new share consumer is
> created after the resume ({{ConditionTimeout}} after 10 seconds waiting for
> the second consumer to subscribe). With the fix it passes. A second test adds
> a {{KafkaShareComponent}} with {{addComponent}} to the started CamelContext
> and consumes with it (passes on main; it guards the {{isStarted()}} check of
> {{startConsumer}}). Unit tests with the fix: camel-kafka-common 11,
> camel-kafka 225 (1 skipped), camel-kafka-share 29, 0 failures.
> *How it was found*: review of the deferred start of the Kafka consumers
> ({{AbstractKafkaComponent.pendingConsumers}}) against the CamelContext
> lifecycle, then the JUnit test above.
> *Affected*: main and 4.23.0 / camel-4.23.x, the release that added
> camel-kafka-share (CAMEL-25029), with the same code (read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
