davsclaus commented on code in PR #28071:
URL: https://github.com/apache/camel/pull/28071#discussion_r4310717013
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java:
##########
@@ -91,6 +91,9 @@ private enum State {
private final BridgeExceptionHandlerToErrorHandler bridge;
private final ReentrantLock lock = new ReentrantLock();
private CommitManager commitManager;
+ // kept while the Kafka consumer is connected: the batching processor
holds the records of earlier polls until its
+ // batch is complete, and the consumer does not return them again
+ private KafkaRecordProcessorFacade recordProcessorFacade;
Review Comment:
Non-blocking: the facade now lives as long as the connection, so a fresh
facade on every `startPolling()` no longer resets a broken
`KafkaRecordBatchingProcessor` by accident. In `processExchange`,
`exchangeList.clear()` runs only after `processBatch(...)` returns. If
`processBatch` ever throws (it catches processor exceptions, so this is
unlikely), the `ArrayBlockingQueue` (capacity `maxPollRecords`) stays full.
Then every later `exchangeList.add(...)` throws `IllegalStateException: Queue
full` until a reconnect. Moving the `clear()` (and the timer restart) into a
`finally` around `processBatch` would close that gap. This could be a follow-up.
--
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]
