如何在SpringKafka监听器上捕捉“经纪人可能不可用”的警告

Question

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

0:20:35.500[TEST_GROUP-0-C-1]警告O.apache.kafka.clients.networkclient-[Consumer clientid=consumer-1,groupid=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

当使用者无法轮询代理时，它将抛出NonResponveConsumerEvent。可以使用@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/#空闲容器

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

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