Spring Cloud Stream Kafka с Confluent не выдает такое же сообщение, как Spring Kafka с Confluent

Question

Я бы хотел использовать Spring Cloud Stream Kafka для своей службы Java / Spring, и мне нужно создавать Confluent сериализованные сообщения, поскольку у меня есть клиенты .NET и NodeJS, которые используют Confluent API для приема моих сообщений.

Итакнасколько мы можем видеть, что Spring Kafka с сериализатором Confluent работает для нас, в то время как Spring Cloud Stream Kafka с сериализатором Confluent доставляет нам проблемы.

Чтобы продемонстрировать, где я вижу разницу в двух случаях, которые я создал2 примера репозиториев на GitHub, содержащих только код, необходимый для создания простого сообщения в обоих случаях.

1. С Spring Kakfa и Confluent https://github.com/donalthurley/springKafkaAvro

2. С SpringCloud Stream Kafka и Confluent https://github.com/donalthurley/springCloudKafkaAvro

Я думаю, что я настроил параметры конфигурации с флагом useNativeEncoding и конфигурацией слитного сериализатора правильно для приложения Spring Cloud, и это можно увидеть в источникекод здесь https://github.com/donalthurley/springCloudKafkaAvro/blob/master/src/main/resources/application.yaml#L8

      kafka:
        binder:
          useNativeEncoding: true
          brokers: 127.0.0.1:9092
        bindings:
          output:
            producer:
              configuration:
                schema.registry.url: http://127.0.0.1:8081
                key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
                value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

Я посылаю такое же простое сообщение отв моем приложении Spring Kafka и в приложении Spring Cloud Stream Kafka отображаются журналы.

Producing Kafka person event: {"lastName": "Doe", "firstName": "John"}

Когда я использую браузер пользовательского интерфейса Kafka Topics из среды докера Kafka, см. https://hub.docker.com/r/landoop/fast-data-dev/, и просмотритев исходных данных сообщения они различаются в обоих случаях.

Для Spring Kafka это выглядит более корректно, поскольку браузер распознает и отображает поля внутри значения сообщения.

[
  {
    "topic": "test_spring_kafka",
    "key": "3197449393600061094",
    "value": {
      "lastName": "Doe",
      "firstName": "John"
    },
    "partition": 0,
    "offset": 0
  }
]

В Spring Cloud StreamКафка необработанных данных браузер не может распознать поля, которые показывают, что сообщения не совпадают.

[
  {
    "topic": "test_spring_cloud_kafka",
    "key": "-6214497758709596999",
    "value": "\u0006Doe\bJohn",
    "partition": 0,
    "offset": 0
  }
]

Я думаю, что может быть проблема при создании сообщений Confluent с использованием Spring Cloud Stream Kafka и что Spring Kafkaреализация производит их правильно, но, может быть, я что-то упустил в моей реализации, и кто-то может помочь мне с этой проблемой?

Answer 1

Проблема в том, как вы настраиваете useNativeEncoding. Это не вступало в силу. Эта конфигурация должна работать:

spring:
  application:
    name: springCloudKafkaAvro
  cloud:
    stream:
      schemaRegistryClient:
        endpoint: http://127.0.0.1:8081
      kafka:
        binder:
            brokers: 127.0.0.1:9092
        bindings:
          output:
            producer:
              configuration:
                schema.registry.url: http://127.0.0.1:8081
                key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
                value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      bindings:
        output:
          destination: test_spring_cloud_kafka
          producer:
            useNativeEncoding: true

Обратите внимание, как useNativeEncoding реорганизован из вашей первоначальной конфигурации.