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

【问题标题】：How to do message filtering at Kafka record level when using SpringCloudStream?使用 SpringCloudStream 时如何在 Kafka 记录级别进行消息过滤？
【发布时间】：2021-05-12 21:42:56
【问题描述】：

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

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

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

我正在考虑配置 RecordFilterStrategy。

【问题讨论】：

标签： apache-kafka spring-kafka spring-cloud-stream

【解决方案1】：

Spring Cloud Stream 不支持RecordFilterStrategy。

您可以添加ListenerContainerCustomizer bean example here 并将RecordInterceptor 添加到侦听器容器中。如果拦截器返回null，则不调用监听器并提交偏移量，就像调用了监听器并正常退出一样。

【讨论】：

感谢很多，这就是我在此期间发现的:)。是否有更低级别的实现方式？我希望尽可能减少对表演的影响。
我不知道。拦截器应该增加很少的开销。