spring kafka：从有效负载中过滤KafkaNull值答案

【问题标题】：spring kafka: filtering KafkaNull values from payloadspring kafka：从有效负载中过滤KafkaNull值
【发布时间】：2021-10-29 08:47:51
【问题描述】：

我的消费者配置如下：

    @Bean
    public Consumer<Message<List<Foo>>> consumer() {
        return message ->  {
            message.getPayload().forEach(it -> {
              // process
            })
        };
    }

我还配置了ErrorHandlingDeserializer，它可能会在消息有效负载集合中产生KafkaNull 值。我无法过滤此类值的问题是因为使用 forEach() 方法访问集合会产生 ClassCastException： java.lang.ClassCastException: class org.springframework.kafka.support.KafkaNull cannot be cast to class Foo

如何从处理中排除 KafkaNull 值（不将使用者的签名更改为 Consumer<Message<List<Object>>>）？

【问题讨论】：

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

【解决方案1】：

所以我们目前没有过滤功能。我确实提出了这个问题 - https://github.com/spring-cloud/spring-cloud-function/issues/736

但现在对你来说最好的方法是做这样的事情：

public Consumer<Message<List<Object>>> consumer() {
        return message ->  {
            message.getPayload().stream().filter(it -> it instanceof Foo).forEach(it -> {
              // process
            })
        };
}

【讨论】：