Слияние Kafka Avro Serializer и Spring Cloud

Question

Я пытаюсь создать событие для Kafka с использованием Spring Cloud и Kafka Avro Serializer.

В моем application.yml у меня есть конфигурация ниже, но когда Serializer пытается сгенерировать сообщение, он генерирует как байты, потому что объект, переданный методу getScheme в KafkaSerializer, является байтовым массивом и не является GenericRecord. Я думаю, что мне нужен конкретный MessageConverter в Spring Cloud, но я его не нахожу.

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

Как я могу это сделать? Как я могу решить это?

Answer 1

См. документацию и, в частности, вам нужно установить свойство производителя useNativeEncoding.

useNativeEncoding

Если установлено значение true, исходящее сообщение сериализуется напрямую клиентской библиотекой, которая должна быть настроена соответствующим образом (например, путем установки соответствующего сериализатора значений производителя Kafka). Когда эта конфигурация используется, сортировка исходящих сообщений не основана на contentType привязки. Когда используется собственное кодирование, потребитель обязан использовать соответствующий декодер (например, десериализатор потребительских значений Kafka) для десериализации входящего сообщения. Кроме того, при использовании встроенного кодирования и декодирования свойство headerMode = embeddedHeaders игнорируется, а заголовки не включаются в сообщение. См. Потребительское свойство useNativeDecoding.