春雲kafka和avro序列化問題

我使用spring-cloud-stream-schema來讀取來自kafka的avro消息。我在MessagesChannels配置輸入通道：春雲kafka和avro序列化問題

@Input("topicName1") 
SubscribableChannel fromInput1();

我有這樣的配置文件：

@Configuration 
@EnableBinding(MessagesChannels.class) 
@EnableSchemaRegistryClient 
public class MessageConfiguration { 
    @Bean 
    public MessageConverter topic1MessageConverter() throws IOException { 
    return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes")); 
    } 
}

我的消費被稱爲與

fromInput1().subscribe(this::onMessage); 

void onMessage(Message message) { 
}

當我實際發送消息我得到這個錯誤：

nested exception is java.lang.ClassCastException: 
org.apache.avro.generic.GenericData$Record cannot be cast to [B

實際上原始字節被正確解析爲org.apache.avro.generic.GenericData$Record。但春天需要Message班。如何將GenericData$Record轉換爲Message或如何將GenericData$Record直接轉換爲由avro-tools類生成？

更多細節：

2017-03-06 11:23:10.695 ERROR 19690 --- [afka-listener-1] o.s.kafka.listener.LoggingErrorHandler : Error while processing: ConsumerRecord(topic = topic1, partition = 0, offset = 7979, CreateTime = 1488784987569, checksum = 623709057, serialized key size = -1, serialized value size = 36, key = null, value = {"foor": "bar"}) 

org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cl[email protected]4bf9d802]; nested exception is java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record cannot be cast to [B 
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:139) 
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:70) 
    at org.springframework.integration.channel.FixedSubscriberChannel.send(FixedSubscriberChannel.java:64)

來源

2017-03-06 ipoteka

我認爲你需要設置contentType爲使用application/*+avro作爲傳入消息通道指定here

來源

2017-03-06 11:44:12

春雲kafka和avro序列化問題

回答

相關問題