How to catch warning “Broker may not be available” at the Spring Kafka Listener

Question

I am working on a POC for implementing a kafka cluster in my project. I have setup a kafka cluster in my local machine with 3 brokers. Now I am sending messages to the Kafka server using Spring MVC REST service which is internally using Spring Kafka to produce and consume messages to and from the Kafka cluster. Now i am trying to send alerts when the consumer is unable to receive messages from the topic when the broker is down. I shutdown the only broker to which the consumer is connected. I am not getting any exception in my logs but i got the following warning messages.

0:20:35.500 [TEST_GROUP-0-C-1] WARN o.apache.kafka.clients.NetworkClient - [Consumer clientId=consumer-1, groupId=TEST_GROUP] Connection to node 2147483645 could not be established. Broker may not be available.

Is it possible to catch this warning message so that i can send alert when my consumer has lost connection? Below is my consumer code.

private static final Logger LOGGER = LoggerFactory.getLogger(ListenerServiceImpl.class);
    @Autowired
    Dao<RnMessage> messageDao;
    @Autowired
    MessageService messageService;

    @KafkaListener(id = "TEST_GROUP", topics = "TESTQUEUE", errorHandler="eventQueueMessageListenerExceptionHandler")
    public void listenMessageInQueue(String msg) {

        try {
            //String str = new String(msg, "UTF-8");
            LOGGER.info("receiving payload='{}'", msg);
            messageDao.saveMessage(msg);
            messageService.sendMessageToOutQueue(msg);
        }catch(Exception e) {
            e.printStackTrace();
        }       
    }

show consumer configs, and also check zookeeper and kafka server are up — Deadpool
With a different setup, you would ship your logs to something like Splunk or Elasticsearch, which are configured with alerting on specific log conditions. Then, you wouldn't be adding boilerplate altering code into all your Kafka clients — OneCricketeer

Shankar P S Shankar P S · Accepted Answer · 2019-01-05T02:54:55

The Consumer will throw a NonResponsiveConsumerEvent when it cannot poll the broker. It is possible to catch this event with an @EventListener.

Here is a sample code:

@EventListener()
public void eventHandler(NonResponsiveConsumerEvent event) {
    //When Kafka server is down, NonResponsiveConsumerEvent error is caught here.
    System.out.println("CAUGHT the event "+ event);
}

You can see more details in the docs. https://docs.spring.io/spring-kafka/reference/htmlsingle/#idle-containers

I am specifically referring to this in the docs.

if the broker is unreachable (at the time of writing), the consumer poll() method does not exit, so no messages are received, and idle events can’t be generated. To solve this issue, the container will publish a NonResponsiveConsumerEvent

A working JUnit test code is here - https://github.com/shankarps/KafkaDemo/blob/master/src/test/java/com/kafkademo/stack/TestConsumerWithNonAvailableBroker.java

How to catch warning “Broker may not be available” at the Spring Kafka Listener

1 Answers