如何检测spring kafka使用者何时停止从1个分区获取消息？

Question

我有3个spring kafka消费者（同一组）从3个分区获得消息。我想检测其中一个使用者何时停止从一个分区读取（其他两个使用者继续从其他两个分区读取）。到目前为止，这种情况已经发生过两次，当检测到时，很容易通过重新启动所有消费者来修复，这将导致重新平衡。问题是在这两种情况下，早点知道会很好。所以我尝试使用ListenerContainerIdleEvent，如下所示-

@EventListener
public void eventHandler(ListenerContainerIdleEvent event) {
    LOG.info("idle event fired! listnerId=" + event.getListenerId());

    Collection<org.apache.kafka.common.TopicPartition> partitions = event.getTopicPartitions();
    partitions.forEach(p ->
            LOG.info("partition: " + p.partition() + " topic:" + p.topic()));
}

这是我的测试结果-

当一个使用者被分配到多个分区时，当没有从一个分区读取消息（无论出于什么原因……使用者问题或没有消息可从分区读取）时，我是否可以得到通知？

更新：03/27/2018

我不确定我是否应该问一个与此相关的新问题，所以试图先扩展这个问题。我有一个消费者消费从1个主题与3个分区。我已经将IdleEventInterval=30secs设置为。每隔30秒，我就会收到以下日志消息。

  @EventListener
public void eventHandler(ListenerContainerIdleEvent event) {
    LOG.info("No messages received for " + event.getIdleTime() + " milliseconds");

Collection<org.apache.kafka.common.TopicPartition> partitions = event.getTopicPartitions();
partitions.forEach(p ->
    LOG.info("partition: " + p.partition() + " topic:" + p.topic()));

}

1）为什么该事件每30秒调用4次？

2）为什么每个消息集的分区信息不一致？有时没有分区信息，有时同一个消息集中的分区重复，等等。

Answer 1

...没有可从分区读取的消息

如果并发度为1，且有3个分区，则所有三个分区将由同一个使用者处理。如果一个使用者被分配到一个以上的主题，并且在一段时间内没有收到来自特定主题的消息，框架中目前没有任何东西可以发布事件。

如果将容器并发性增加到3，则将有3个使用者-每个分区一个。如果事件处于空闲状态，则每个使用者将发布该事件。监听器必须是线程安全的，因为将有3个线程调用它，而且通常是并发的。

如果一个主题中有消息，而该主题被分配给的使用者没有接收到这些消息，但该使用者却在积极地从其他分区接收消息，这将是相当奇怪的。这需要Kafka家族的帮助。