shashank created CAMEL-25872:
--------------------------------
Summary: 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
{{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 keeps it
pending until the notification. 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}} ({{MockShareConsumer}}
based, as {{KafkaShareConsumerTest}}): 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. Unit
tests with the fix: camel-kafka-common 11, camel-kafka 225 (1 skipped),
camel-kafka-share 28, 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
(v8.20.10#820010)
