Spring Kafka - Consumers Detecting Broker Unavailable

[yihfenglow]

I understand from https://stackoverflow.com/questions/47524315/handle-failure-in-case-the-kafka-broker-is-down that since 1.3.2, there is a "NonResponsiveConsumerEvent" that triggers when the broker is unavailable.

When trying this, I note that the event is triggered regardless of whether the consumer polling is responsive or not.

Digging deeper into the code, I note that the "last" variable in KafkaMessageListenerContainer.java isn't being updated in the processCommits() method if ackMode = manual, batch, record or count. Because the "last" variable isn't being updated, it looks like the checkConsumer() method in KafkaMessageListenerContainer.java will always trigger the NonResponsiveConsumerEvent.

Would you please be able to take a look to see if this is a bug?

Thanks!

[yihfenglow]

After a little more investigation, it looks like the issue is that the "last" variable is meant to capture the last read/commit.

What is more apt is a new lastSuccessfulPoll variable that should be updated to System.currentTimeMillis() after each call to this.consumer.poll() in KafkaMessageListenerContainer.

[garyrussell]

Yes; this is a bug; thanks; we can't update last for each poll because it's used for time-based commits, but it's definitely a bug.

[AleksandarTokarev]

@garyrussell NonResponsiveConsumerEvent does not seem to work for me in Spring Boot 2.7.8 and Spring Kafka 2.8.1. Anything I am missing? My idea is to survive Kafka Broker outage - by implementing a custom health check which would return Status.DOWN

[garyrussell]

I no longer work on this project.

That event is obsolete (this issue is nearly 6 years old); the event was used with a much older kafka-clients library, which needed a different threading model, and the consumer thread could be stuck in poll(). This is no longer the case.

Unfortunately, the consumer API provides no mechanism to check broker liveness. You could use something like AdminClient.describeCluster() to check the status.