How to set avro content type on kafka producer with spring boot and spring cloud stream

Question

I am trying to build a working example with avro message producer, kafka and avro consumer using spring boot, spring cloud stream schema registry, kafka binder. The following error occurs when consuming messages:

ERROR 7059 --- [afka-listener-1] o.s.c.s.b.k.KafkaMessageChannelBinder : Unable to deserialize [io.igx.android.Sensor] using the contentType [application/x-java-object;type=io.igx.android.Sensor] java.lang.NullPointerException

I have spring.cloud.stream.bindings.output.contentType=application/+avro on consumer side and spring.cloud.stream.bindings.output.contentType=application/+avro on producer side.

Why is the consumer not seeing and using avro content type instead application/x-java-object.

Answer 1

I think you need the content type application/*+avro at the producer side. (note the * prefix). Once that is set for the producer, Can you confirm if you still need to specify the input contentType for the consumer. I think the consumer can deduce it based on the content type set by the producer.

Answer 2

Channel name was wrong. So problem was I was using spring.cloud.stream.bindings.input.contentType=application/*‌​+avro

instead of this

spring.cloud.stream.bindings.{MY CHANNEL NAME}.contentType=application/*‌​+avro.

When I changed {MY CHANNEL NAME} to the name that i used magic happened.

It works even if i did not specify content type on consumer side.

Also, I need to specify spring.cloud.stream.schemaRegistryClient.endpoint if it's not default, like in my case, both, on producer and consumer side.