RecordFilterStrategy配合Kafka消息监听器使用不生效该如何处理？

根本原因 #

Spring Kafka的ConcurrentMessageListenerContainer不会自动识别监听器实现的RecordFilterStrategy接口，必须手动为监听容器配置过滤策略实例，自定义的filter方法才会被触发。

修正步骤 #

1. 修正配置类代码，手动为容器设置过滤策略，同时修正原代码中的拼写错误

// KafkaConfig修正后代码
ConcurrentMessageListenerContainer<String,Object> kafkaListnerContainer(){
    ContainerProperties prop = new ContainerProperties("topicName");
    // 原代码此处方法名拼写错误，应为setMessageListener，少了字母e
    prop.setMessageListener(myMessageListener);
    ConcurrentMessageListenerContainer<String,Object> lContainer = new ConcurrentMessageListenerContainer(consumerFactory(),prop);
    // 新增以下配置，手动注入过滤策略
    lContainer.setRecordFilterStrategy((RecordFilterStrategy<String, Object>) myMessageListener);
    // 可选配置：被过滤的消息不提交offset，默认false即过滤消息也会提交offset
    // lContainer.setAckDiscarded(true);
    return lContainer ;    
}

2. 检查过滤逻辑判断方向
   filter方法返回true代表当前消息会被丢弃，不会进入onMessage执行业务逻辑；返回false代表消息正常放行，不要写反判断条件。

额外注意 #

原代码中MessageListner、setMessageListner两处存在拼写错误，正确拼写为MessageListener、setMessageListener，拼写错误会导致功能异常，请先修正。

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