如何在Spring Kafka Listener上捕捉“Broker可能不可用”的警告

Question

我正在研究一个POC，用于在我的项目中实现Kafka集群。我已经设置了一个Kafka集群在我的本地机器与3经纪人。现在，我使用Spring MVC REST服务向Kafka服务器发送消息，该服务在内部使用Spring Kafka生成和使用往返于Kafka集群的消息。现在，我试图发送警报时，消费者无法接收消息从主题时，经纪人是下降。我关闭消费者连接到的唯一代理。我没有得到任何异常在我的日志，但我得到了以下警告信息。

0:20:35.500[TEST_GROUP-0-C-1]WARNo.apache.kafka.clients.NetworkClient-[消费者clientId=消费者-1， group pId=TEST_GROUP]无法建立到节点2147483645的连接。经纪人可能不可用。

是否可以捕获此警告消息，以便在消费者失去连接时发送警报？下面是我的消费代码。

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();
        }       
    }

Answer 1

当消费者无法轮询代理时，它将抛出一个Non响应消费者事件。可以使用@EventListener捕获此事件。

下面是一个示例代码：

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

你可以在文档中看到更多细节。https://docs.spring.io/spring-kafka/reference/htmlsingle/#idle-containers

我在文件中特别提到这一点。

如果代理不可达（在撰写本文时），则不退出消费者投票（）方法，因此不会收到任何消息，也无法生成空闲事件。为了解决这个问题，容器将发布一个Non响应消费者事件

JUnit测试代码正在运行https://github.com/shankarps/KafkaDemo/blob/master/src/test/java/com/kafkademo/stack/TestConsumerWithNonAvailableBroker.java