m1a2st commented on code in PR #17440:
URL: https://github.com/apache/kafka/pull/17440#discussion_r1794731648
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1594,7 +1594,11 @@ private boolean updateFetchPositions(final Timer timer) {
try {
CheckAndUpdatePositionsEvent checkAndUpdatePositionsEvent = new
CheckAndUpdatePositionsEvent(calculateDeadlineMs(timer));
wakeupTrigger.setActiveTask(checkAndUpdatePositionsEvent.future());
- cachedSubscriptionHasAllFetchPositions =
applicationEventHandler.addAndGet(checkAndUpdatePositionsEvent);
+ applicationEventHandler.add(checkAndUpdatePositionsEvent);
Review Comment:
but there are another issue in close the consumer, I didn't have good idea
to solve this.
```
[2024-10-10 14:08:59,259] WARN [NodeToControllerChannelManager id=1000
name=registration] Attempting to close NetworkClient that has already been
closed. (org.apache.kafka.clients.NetworkClient:694)
[2024-10-10 14:08:59,261] ERROR [QuorumController id=1000] Cancelling
deferred write event maybeFenceReplicas because the event queue is now closed.
(org.apache.kafka.controller.QuorumController:1184)
Failed to close kafka consumer
org.apache.kafka.common.KafkaException: Failed to close kafka consumer
at
org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer.close(AsyncKafkaConsumer.java:1249)
```
--
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]
