apache-kafka - 使用 SpringCloudStream 时如何在 Kafka 记录级别进行消息过滤？

Question

我正在使用 Spring Cloud Stream (SCS) 和 Kafka 作为活页夹。

我想根据 Kafka 标头对记录进行低级过滤。推荐的方法是什么？

过滤后的消息应该被忽略，偏移量应该被提交。

我正在考虑配置 RecordFilterStrategy。

标签： apache-kafkaspring-kafkaspring-cloud-stream

Answer 1

RecordFilterStrategySpring Cloud Stream 不支持。

您可以在此处添加ListenerContainerCustomizerbean示例并将 a 添加RecordInterceptor到侦听器容器中。如果拦截器返回null，则不调用监听器并提交偏移量，就像调用了监听器并正常退出一样。