davsclaus opened a new pull request, #28060:
URL: https://github.com/apache/camel/pull/28060
CAMEL-25753: a Kafka consumer stopped only after its current poll returned.
`KafkaFetchRecords` held its lock for the whole poll loop, and `stop()` waited
for that lock before it called `consumer.wakeup()`, so the wakeup never ended a
poll.
- **Each consumer** needed up to `pollTimeoutMs` (5 s by default) to stop,
and the routes stop one after another.
- **`camel run --dev`** stops the routes of a reload within 5 s. A reload
with two or more Kafka consumer routes was forced, and the interrupted Kafka
clients logged errors, such as "Failed to close coordinator" and "Error closing
the Kafka consumer: java.lang.InterruptedException".
- **In production,** an application with many Kafka routes was slow to stop.
## What
- **The lock covers the processing of the polled records only, not the
poll.** `stop()` still waits for a processing to finish, including its commit,
up to `shutdownTimeout`, so a wakeup does not interrupt it. Then it wakes up a
consumer that is in poll, and the poll ends at once with the `WakeupException`
the loop already treats as the stop.
- **No wakeup is left pending for records already polled.** A `polling` flag
tells `stop()` whether the fetcher thread is in poll. A consumer between two
polls is not woken up: it sees the stop before it polls again, so a pending
wakeup does not interrupt the records it already has. The fetcher checks the
stop again once the flag is set, so a stop that begins just before a poll is
not missed. As before, a processing that does not finish within the timeout is
woken up regardless.
- **`safeUnsubscribe` treats a `WakeupException` as part of the stop.** It
is logged at debug, and the consumer is closed next.
## Tests
- **`KafkaConsumerStopTest` (new),** with a `MockConsumer` whose poll waits
like a real consumer on an idle topic:
- **Stop during a 10 s poll:** 0.24 s with the fix. On the code before the
fix the test fails: "The route stopped after 9905 ms".
- **Stop during the processing of a record:** the stop waits for it, and
the record reaches the end of the route.
- **All camel-kafka tests pass.** That is 222 unit tests, and 103 + 25
integration tests against a Kafka container: the commit managers, batching,
breakOnFirstError, pause, transactions and the existing stop test.
- **A throwaway integration test** (not committed) stopped three consumer
routes on a real broker, with the default poll timeout and a 5 s shutdown
timeout as in dev mode:
- before: 8.5 s, and the shutdown was forced,
- with the fix: 1.6 s, graceful.
The AI benchmark's kafka-orders project sets
`camel.component.kafka.pollTimeoutMs=500` to work around this. It can drop that
once this is in.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m
--
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]
