Spring-Kafka BatchInterceptor未生效及批量消息过滤方案咨询

Spring Cloud Stream Kafka Binder在批量消费模式下，并不会直接使用Spring Kafka容器BatchInterceptor返回的过滤后ConsumerRecords来构建最终传递给消费者的消息列表。Binder内部会独立处理原始ConsumerRecords的转换逻辑，导致BatchInterceptor的过滤结果被忽略，这就是为什么你看到日志显示过滤成功，但被过滤的记录仍然进入消费者的原因。

针对批量模式下按Header过滤消息的需求，推荐以下几种可行方案：

方案1：在消费者方法中直接过滤（最简单） #

如果你的消费者接收的是List<Message<String>>类型的批量消息，可以直接在消费逻辑中过滤掉不包含指定Header的记录：

@Bean
public Consumer<List<Message<String>>> input() {
    return messages -> {
        List<Message<String>> filteredMessages = messages.stream()
                .filter(msg -> msg.getHeaders().containsKey("TEST"))
                .collect(Collectors.toList());
        
        log.info("Filtered message count: {}", filteredMessages.size());
        filteredMessages.forEach(msg -> log.info("Received payload: {}", msg.getPayload()));
    };
}

如果之前配置的是接收List<String>，需要调整消费绑定的类型为List<Message<String>>，确保Header信息被传递到消费者中。

方案2：自定义BatchMessageConverter（推荐，容器层面过滤） #

通过自定义消息转换器，在ConsumerRecords转换为批量消息的阶段完成过滤，确保只有符合条件的记录被传递给消费者：

@Bean
public BatchMessageConverter filteredBatchMessageConverter() {
    return new BatchMessagingMessageConverter() {
        @Override
        public Message<?> toMessage(ConsumerRecords<?, ?> records, Acknowledgment acknowledgment, Consumer<?, ?> consumer, Type type) {
            // 过滤掉不包含TEST Header的记录
            Map<TopicPartition, List<ConsumerRecord<?, ?>>> filteredRecordMap = records.partitions().stream()
                    .collect(Collectors.toMap(
                            Function.identity(),
                            partition -> records.records(partition).stream()
                                    .filter(record -> Objects.nonNull(record.headers().lastHeader("TEST")))
                                    .collect(Collectors.toList())
                    ));
            ConsumerRecords<?, ?> filteredRecords = new ConsumerRecords<>(filteredRecordMap);
            
            // 调用父类方法完成消息转换
            return super.toMessage(filteredRecords, acknowledgment, consumer, type);
        }
    };
}

然后在配置文件中指定该转换器：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          input-in-0: # 替换为你的消费绑定名称
            consumer:
              batch-mode: true
              message-converter: filteredBatchMessageConverter

方案3：使用单条记录拦截器RecordInterceptor #

如果不需要批量级别的过滤逻辑，可以使用Spring Kafka的RecordInterceptor逐个过滤记录，批量模式下同样生效：

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<String, String>> containerCustomizer() {
    return (container, destinationName, group) -> {
        container.setRecordInterceptor((record, consumer) -> {
            // 返回null表示过滤该条记录
            return Objects.nonNull(record.headers().lastHeader("TEST")) ? record : null;
        });
    };
}

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