问题描述
在将我们的一项服务升级到 spring-cloud-stream
2.0.0.RC3 时,我们在尝试使用由使用旧版本 spring-cloud-stream
- Ditmars.RELEASE:
错误 31241 --- [container-4-C-1] osintegration.handler.LoggingHandler:org.springframework.messaging.converter.MessageConversionException:无法从 [[B] 转换为 [com.watercorp.messaging.types].incoming.UsersDeletedMessage] 用于 GenericMessage [payload=byte[371], headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@62029d0d, kafka_timestampType=CREATE_176451d40d=CREATE_17645150022e6-b629-229b-4fa961c73f2d, type=USERS_DELETED, kafka_receivedPartitionId=4, contentType=text/plain, kafka_receivedTopic=user, kafka_receivedTimestamp=15216421677sMessage headerskafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@62029d0d, kafka_timestampType=CREATE_TIME, message_id=1645508761, id=f4e947de-22e947de-22e947de-22e947de-22e947de-29e6-29dtype_dkafka_timestampType=CREATE_TIME, 类型=29e6-29fkafkafkafka-29e6-2262261dkafka_timestampType=c29e6-2262d1dkafka_timestampType=c29e6-b62d文本/普通,kafka_receivedTopic=用户,kafka_receivedTimestamp=1521641760698,时间戳=1521641772477}]在 org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:144)在 org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:116)在 org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:137)在 org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:109)在 org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55)在 org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)在 org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:164)在 org.springframework.cloud.stream.binding.DispatchingStreamListenerMessageHandler.handleRequestMessage(DispatchingStreamListenerMessageHandler.java:87)在 org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)在 org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:157)在 org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)在 org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:132)在 org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105)在 org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:463)在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:407)在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181)在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160)在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)在 org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)在 org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203)在 org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:70)在 org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:387)在 org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:364)在 org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1001)在 org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:981)在 org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:932)在 org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:801)在 org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:689)在 java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)在 java.util.concurrent.FutureTask.run(FutureTask.java:266)在 java.lang.Thread.run(Thread.java:745)
看起来原因是与消息一起发送的 消费者配置: 消费者@StreamListener: 这是 RC3 中的一个错误;最近在 master 上修复;它将在本月底到期的 GA 版本中发布.同时,您可以尝试使用 2.0.0.BUILD-SNAPSHOT 吗? 我能够重现该问题并使用快照为我修复了它... 编辑 为了完整性: Ditmars 制作人 和 埃尔姆赫斯特消费者: 和 结果: While upgrading one of our services to It looks like that the cause is that the Consumer configuration: Consumer @StreamListener: This is a bug in RC3; recently fixed on master; it will be in the GA release due at the end of the month. In the meantime, can you try with 2.0.0.BUILD-SNAPSHOT? I was able to reproduce the problem and using the snapshot fixed it for me... EDIT For completeness: Ditmars producer and Elmhurst consumer: and Result: The 这篇关于spring-cloud-stream 消息转换异常的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!contentType
标头是 text/plain
虽然它应该是 application/json代码>.
生产者配置:弹簧:云:流:卡夫卡:粘合剂:经纪人:kafka默认代理端口:9092zkNodes:动物园管理员默认ZkPort:2181最小分区数:2复制因子:1自动创建主题:true自动添加分区:真标头:类型,message_idrequiredAcks:1配置:[security.protocol]":PLAINTEXT #TODO:这是一种解决方法.应该是 security.protocol绑定:用户输出:制作人:同步:真配置:重试:10000默认:粘合剂:卡夫卡内容类型:应用程序/json组:用户服务消费者:最大尝试:1制作人:partitionKeyExtractorClass: com.watercorp.user_service.messaging.PartitionKeyExtractor绑定:用户输出:目的地:用户制作人:分区数:5 <groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream</artifactId><version>2.0.0.BUILD-SNAPSHOT</version></依赖><依赖><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-kafka</artifactId><version>2.0.0.BUILD-SNAPSHOT</version><排除事项><排除><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-kafka-core</artifactId></排除></排除项></依赖><依赖><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-kafka-core</artifactId><version>2.0.0.BUILD-SNAPSHOT</version></依赖>
@SpringBootApplication@EnableBinding(Source.class)公共类 So49409104Application {公共静态无效主(字符串 [] args){SpringApplication.run(So49409104Application.class, args);}@豆公共 ApplicationRunner 运行程序(MessageChannel 输出){返回参数 ->{foo foo = new foo();foo.setBar("bar");output.send(new GenericMessage<>(foo));};}公共静态类 Foo {私人字符串栏;公共字符串 getBar() {返回 this.bar;}public void setBar(String bar) {this.bar = 酒吧;}@覆盖公共字符串 toString() {return "Foo [bar=" + this.bar + "]";}}}
弹簧:云:流:绑定:输出:目的地:so49409104a内容类型:应用程序/json制作人:标头模式:embeddedHeaders
@SpringBootApplication@EnableBinding(Sink.class)公共类 So494091041Application {公共静态无效主(字符串 [] args){SpringApplication.run(So494091041Application.class, args);}@StreamListener(Sink.INPUT)公共无效听(Foo foo){System.out.println(foo);}公共静态类 Foo {私人字符串栏;公共字符串 getBar() {返回 this.bar;}public void setBar(String bar) {this.bar = 酒吧;}@覆盖公共字符串 toString() {return "Foo [bar=" + this.bar + "]";}}}
弹簧:云:流:绑定:输入:群:so49409104目的地:so49409104a消费者:标头模式:embeddedHeaders内容类型:应用程序/json
Foo [bar=bar]
header-mode
是必需的,因为 2.0 中的默认值是 native
,现在 Kafka 支持本地头.spring-cloud-stream
2.0.0.RC3 we encountered an exception when trying to consume a message that was produced by a service that uses older version of spring-cloud-stream
- Ditmars.RELEASE:contentType
header that is sent with the message is text/plain
although it should be application/json
.
The producer configuration:
spring:
cloud:
stream:
kafka:
binder:
brokers: kafka
defaultBrokerPort: 9092
zkNodes: zookeeper
defaultZkPort: 2181
minPartitionCount: 2
replicationFactor: 1
autoCreateTopics: true
autoAddPartitions: true
headers: type,message_id
requiredAcks: 1
configuration:
"[security.protocol]": PLAINTEXT #TODO: This is a workaround. Should be security.protocol
bindings:
user-output:
producer:
sync: true
configuration:
retries: 10000
default:
binder: kafka
contentType: application/json
group: user-service
consumer:
maxAttempts: 1
producer:
partitionKeyExtractorClass: com.watercorp.user_service.messaging.PartitionKeyExtractor
bindings:
user-output:
destination: user
producer:
partitionCount: 5
spring:
cloud:
stream:
kafka:
binder:
brokers: kafka
defaultBrokerPort: 9092
minPartitionCount: 2
replicationFactor: 1
autoCreateTopics: true
autoAddPartitions: true
headers: type,message_id
requiredAcks: 1
configuration:
"[security.protocol]": PLAINTEXT #TODO: This is a workaround. Should be security.protocol
bindings:
user-input:
consumer:
autoRebalanceEnabled: true
autoCommitOnError: true
enableDlq: true
default:
binder: kafka
contentType: application/json
group: enrollment-service
consumer:
maxAttempts: 1
headerMode: embeddedHeaders
producer:
partitionKeyExtractorClass: com.watercorp.messaging.PartitionKeyExtractor
headerMode: embeddedHeaders
bindings:
user-input:
destination: user
consumer:
concurrency: 5
partitioned: true
@StreamListener(target = UserInput.INPUT, condition = "headers['type']=='" + USERS_DELETED + "'")
public void handleUsersDeletedMessage(@Valid UsersDeletedMessage usersDeletedMessage, @Header(value = "kafka_receivedPartitionId",
required = false) String partitionId, @Header(value = KAFKA_TOPIC_HEADER_NAME, required = false) String topic, @Header(MESSAGE_ID_HEADER_NAME) String messageId) throws Throwable {
logger.info(String.format("Received users deleted message message, message id: %s topic: %s partition: %s", messageId, topic, partitionId));
handleMessageWithRetry(_usersDeletedMessageHandler, usersDeletedMessage, messageId, topic);
}
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
<exclusions>
<exclusion>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
</dependency>
@SpringBootApplication
@EnableBinding(Source.class)
public class So49409104Application {
public static void main(String[] args) {
SpringApplication.run(So49409104Application.class, args);
}
@Bean
public ApplicationRunner runner(MessageChannel output) {
return args -> {
Foo foo = new Foo();
foo.setBar("bar");
output.send(new GenericMessage<>(foo));
};
}
public static class Foo {
private String bar;
public String getBar() {
return this.bar;
}
public void setBar(String bar) {
this.bar = bar;
}
@Override
public String toString() {
return "Foo [bar=" + this.bar + "]";
}
}
}
spring:
cloud:
stream:
bindings:
output:
destination: so49409104a
content-type: application/json
producer:
header-mode: embeddedHeaders
@SpringBootApplication
@EnableBinding(Sink.class)
public class So494091041Application {
public static void main(String[] args) {
SpringApplication.run(So494091041Application.class, args);
}
@StreamListener(Sink.INPUT)
public void listen(Foo foo) {
System.out.println(foo);
}
public static class Foo {
private String bar;
public String getBar() {
return this.bar;
}
public void setBar(String bar) {
this.bar = bar;
}
@Override
public String toString() {
return "Foo [bar=" + this.bar + "]";
}
}
}
spring:
cloud:
stream:
bindings:
input:
group: so49409104
destination: so49409104a
consumer:
header-mode: embeddedHeaders
content-type: application/json
Foo [bar=bar]
header-mode
is required since the default in 2.0 is native
now that Kafka supports headers natively.