CODEWITHSUNDEEP
javaapache-kafkaspring-cloudavrospring-cloud-stream
Nikita
Nikita

Reputation: 4515

Spring cloud kafka and avro serialization issue

I use spring-cloud-stream-schema to read avro messages from kafka. I configured input channel in MessagesChannels:

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

I have configuration file like that:

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

And my consumer is called with 

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

void onMessage(Message message) {
}

When I actually sent message I got this error:

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

Actually raw bytes are parsed correctly into org.apache.avro.generic.GenericData$Record. But spring requires Message class. How to cast GenericData$Record to Message or how to cast GenericData$Record directly to generated by avro-tools class?

More details:

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.cloud.stream.binder.AbstractMessageChannelBinder$ReceivingHandler@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)

Upvotes: 0

Views: 901

Answers (1)

Ilayaperumal Gopinathan
Ilayaperumal Gopinathan

Reputation: 4179

I think you need to set the contentType for the incoming message channel to use application/*+avro as specified here

Upvotes: 1

Related Questions