Claus Ibsen created CAMEL-25753:
-----------------------------------
Summary: camel-kafka - stop the consumer without waiting for its
current poll, so a reload with several Kafka routes is fast and logs no errors
Key: CAMEL-25753
URL: https://issues.apache.org/jira/browse/CAMEL-25753
Project: Camel
Issue Type: Improvement
Components: camel-kafka
Reporter: Claus Ibsen
Fix For: 4.24.0
A Kafka consumer stops only after its current poll returns.
KafkaFetchRecords.startPolling() takes the lock before the poll loop and
releases it after the loop. safeStop() waits for that lock (up to
shutdownTimeout) and only then calls consumer.wakeup(). So the wakeup, which is
meant to break a blocking poll, comes after the poll has returned. Pausing the
consumer (the suspend phase of a graceful shutdown) waits for the poll in the
same way. Each consumer needs up to pollTimeoutMs (5000 ms by default) to pause
and to stop, and the routes stop one after another.
With camel run --dev, the shutdown timeout is 5 seconds (Run.java). A reload
with two or more Kafka consumer routes goes past it:
{noformat}
WARN DefaultShutdownStrategy : Timeout occurred during graceful shutdown.
Forcing the routes to be shutdown now. Notice: some resources may still be
running as graceful shutdown did not complete successfully.
ERROR ClassicKafkaConsumer : [Consumer clientId=consumer-..., groupId=...]
Failed to close coordinator with a timeout(ms)=30000
ERROR KafkaProducer : [Producer clientId=producer-4] Interrupted while joining
ioThread
ERROR KafkaFetchRecords : Error closing the Kafka consumer:
java.lang.InterruptedException (this error will be ignored)
{noformat}
These errors come on every such reload, while nothing is wrong with the routes.
In production (shutdown timeout 45 s), an application with many Kafka consumer
routes is slow to stop.
Measured with three consumer routes on a local broker: the first consumer took
2.6 s to stop, and the second was still stopping when the 5 s ran out.
Workaround: a lower poll timeout, for example
camel.component.kafka.pollTimeoutMs=500.
Proposal: wake the consumer when the stop (and the pause) begins. wakeup() is
thread safe and only affects a blocking call. The poll loop then treats the
WakeupException as the stop.
Found while adding the kafka-orders example to the AI benchmark.
_Claude Code on behalf of davsclaus_
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
