Confluent Kafka Avro Serializer and Spring Cloud

Question

I'm trying to produce an event on Kafka using Spring Cloud and Kafka Avro Serializer.

In my application.yml I have the configuration below, but when the Serializer tries to generate the message, It generates as bytes, because the object passed to getScheme method in KafkaSerializer is a byte array and it isn't a GenericRecord. I think that I need a specific MessageConverter in Spring Cloud, but I don't find any.

cloud:
stream:
  kafka:
    binder:
      brokers:
        - 'localhost:9092'
      useNativeDecoding: true
    bindings:
      Ptr-output:
          producer:
            configuration:
              schema.registry.url: 'http://localhost:8081'
              key.serializer: org.apache.kafka.common.serialization.StringSerializer
              value.serializer: com.abc.message.ptr.KafkaSerializer
  schemaRegistryClient:
    endpoint: 'http://localhost:8081'
  bindings:
     Ptr-output:
      contentType: application/*+avro
      destination: Ptr
  schema:
    avro:
      schema-locations: 'classpath:avro/Ptr.avsc'
      dynamic-schema-generation-enabled: false

How can I do? How can I solve it?

Gary Russell Gary Russell · Accepted Answer · 2018-10-31T18:25:55

See the documentation and, in particular, you need to set the producer property useNativeEncoding.

useNativeEncoding

When set to true, the outbound message is serialized directly by the client library, which must be configured correspondingly (for example, setting an appropriate Kafka producer value serializer). When this configuration is being used, the outbound message marshalling is not based on the contentType of the binding. When native encoding is used, it is the responsibility of the consumer to use an appropriate decoder (for example, the Kafka consumer value de-serializer) to deserialize the inbound message. Also, when native encoding and decoding is used, the headerMode=embeddedHeaders property is ignored and headers are not embedded in the message. See the consumer property useNativeDecoding.

Confluent Kafka Avro Serializer and Spring Cloud

2 Answers