[
https://issues.apache.org/jira/browse/CAMEL-25872?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25872:
-----------------------------
Description:
{{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._
was:
{{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._
> 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
>
> {{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
(v8.20.10#820010)
