Spring Cloud Stream如何获取ListenerContainerIdleEvent这类KafkaEvent？

为啥收不到ListenerContainerIdleEvent？ #

你用的是Kafka Streams绑定器，它不会发布普通Kafka消费者绑定器的ListenerContainerIdleEvent——这类事件是给传统单消费者用的。Kafka Streams绑定器有自己的专属空闲事件：KafkaStreamsIdleEvent。

正确监听空闲事件的代码 #

把原来的监听逻辑改成监听KafkaStreamsIdleEvent就行：

import org.springframework.cloud.stream.binder.kafka.streams.event.KafkaStreamsIdleEvent;
import org.springframework.context.event.EventListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// 在你的业务类或配置类里添加这个方法
private static final Logger log = LoggerFactory.getLogger(YourClass.class);

@EventListener
public void onKafkaStreamsIdle(KafkaStreamsIdleEvent event) {
    log.info("Kafka Streams 拓扑 {} 已空闲 {} 毫秒",
             event.getKafkaStreams().toString(),
             event.getIdleTimeDuration().toMillis());
    // 这里写入空闲状态下的业务逻辑
}

配置要调整到位 #

看你的application.yaml，有些配置属于冗余或错配，idle-event-interval是Kafka Streams专属参数，不需要在普通消费者绑定配置里设置，调整后更清晰：

spring:
  cloud:
    function:
      definition: codeObject
    stream:
      events:
        enabled: true # 必须开启事件发布，你已配置无需修改
      kafka.streams:
        binder:
          application-id: app-id
          deserializationExceptionHandler: logAndFail
          configuration:
            default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            default.value.serde: io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde
            schema.registry.url: xxx
            auto.offset.reset: earliest
            idle-event-interval: 5000 # 全局所有Kafka Streams拓扑的空闲检测间隔
          functions:
            codeObject:
              application-id: input-topic.v1
              configuration:
                idle-event-interval: 5000 # 单独给该函数设置间隔，会覆盖全局配置
        bindings:
          codeObject-in-0:
            consumer:
              destination-is-pattern: false
      bindings:
        codeObject-in-0:
          destination: input-topic.v1
          consumer:
            auto-startup: true

注意：可以删掉bindings.codeObject-in-0.consumer下的idle-event-interval，那是给普通消费者用的，对Kafka Streams无效。

想判断消费滞后为0？这么做 #

如果要确认消费已经追上最新偏移量（滞后为0），仅靠空闲事件不够，得主动对比偏移量：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListOffsetsResult;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsOperations;

import java.util.List;
import java.util.stream.Collectors;

@Autowired
private KafkaStreamsOperations<String, CodeObject> kafkaStreamsOperations;

@Autowired
private AdminClient adminClient;

public boolean isConsumptionUpToDate(String topic) {
    KafkaStreams streams = kafkaStreamsOperations.getKafkaStreams();
    if (streams == null || streams.state() != KafkaStreams.State.RUNNING) {
        return false;
    }

    // 获取目标topic的所有分区
    List<TopicPartition> partitions = streams.metadataForTopics(topic).values().stream()
            .flatMap(md -> md.partitions().stream())
            .map(p -> new TopicPartition(topic, p))
            .collect(Collectors.toList());

    // 获取每个分区的最新偏移量
    ListOffsetsResult latestOffsets = adminClient.listOffsets(
            partitions.stream().collect(Collectors.toMap(p -> p, p -> ListOffsetsResult.ListOffsetsRequestEarliestOrLatest.LATEST))
    );

    // 对比消费者当前偏移量与最新偏移量
    for (TopicPartition partition : partitions) {
        // 替换为你拓扑中实际的状态存储名称
        long currentOffset = streams.store("your-state-store-name", QueryableStoreTypes.keyValueStore())
                .getPartitionOffset(partition);
        long latestOffset = latestOffsets.partitionResult(partition).join().offset();
        // 减1是因为latestOffset是下一条待消费消息的位置
        if (currentOffset < latestOffset - 1) {
            return false;
        }
    }
    return true;
}

注意：需要提前注入AdminClient，且你的Kafka Streams拓扑中必须包含可查询的状态存储（比如通过toTable()或store()创建的存储）。

内容的提问来源于stack exchange，提问作者manuelgr