带条件的streamlistener

h43kikqp 于 2021-06-08 发布在 Kafka

关注(0)|答案(2)|浏览(427)

我正在尝试使用streamlistener注解和condition attirbute创建使用者。但是，出现以下异常：
org.springframework.core.convert.conversionfailedexception:未能将值“test”的类型[java.lang.string]转换为类型[java.lang.integer]；嵌套异常为java.lang.numberformatexception:对于输入字符串：“test”
测试侦听器：

@StreamListener(target=ITestSink.CHANNEL_NAME,condition="payload['test'] == 'test'")
public void test(@Payload TestObj message) {
    log.info("message is {}",message.getName());
}

测试对象：

@Data
@ToString(callSuper=true)
public class TestObj {

    @JsonProperty("test")
    private String test;

    @JsonProperty("name")
    private String name;

}

有人能帮助解决这个问题吗？

apache-kafka spring-cloud-stream

来源：https://stackoverflow.com/questions/51541819/spring-cloud-stream-kafka-2-0-streamlistener-with-condition

2条答案

按热度按时间

hc8w905p1#

消息的有效负载尚未从有线格式（字节[]）转换为所需类型。换句话说，它还没有经历内容类型协商中描述的类型转换过程。
因此，除非使用计算原始数据的spel表达式（例如，字节数组中第一个字节的值），否则请使用基于消息头的表达式（例如condition=“headers['type']=='dog'”。
例子：

@StreamListener(target = Sink.INPUT, condition = "headers['type']=='bogey'")
    public void receiveBogey(@Payload BogeyPojo bogeyPojo) {
       // handle the message
    }

请查看此处的spring文档。

赞(0）回复(0）举报 2021-06-08

44u64gxh2#

从你的表现来看，它应该有用。我建议您删除条件，然后将断点设置为debug。然后你就可以知道实际的类型是什么了。

赞(0）回复(0）举报 2021-06-08