有汇流的SpringCould SteamKafka与有汇流的SpringKafka不产生相同的信息

Question

我希望将Spring Cloud Stream Kafka用于我的Java/Spring服务，并且我需要生成汇流序列化消息，因为我有使用汇流API来使用我的消息的.NET和NodeJS客户端。

就我们所见，使用汇流序列化器的Spring Kafka正在为我们工作，而使用汇流序列化器的Spring Cloud Stream Kafka正在给我们带来问题。

为了演示这两种情况下的区别，我在GitHub上创建了两个示例存储库，其中只包含在这两种情况下生成简单消息所需的代码。

>

使用Spring Kakfa和Confluent https://github.com/donalthurley/SpringKafkaavro

使用Spring Cloud Stream Kafka和Confluent https://github.com/donalthurley/SpringCloudKafkaavro

我认为我已经为Spring Cloud应用程序正确地配置了带有UseNativeEncode标志的配置设置和汇合序列化器配置，这些配置可以在源代码中看到，这里有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"}

当我从我的docker Kafka环境中使用Kafka Topics UI浏览器时，请参阅https://hub.docker.com/r/landoop/fast-data-dev/，并查看消息raw data。

当浏览器识别并显示消息值内的字段时，Spring Kafka看起来更正确。

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

在Spring Cloud流Kafka原始数据中，浏览器无法识别字段，这表明消息不是相同的。

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

我认为使用Spring Cloud Stream Kafka生成汇合消息可能存在问题，而且Spring Kafka实现正确地生成了这些消息，但也许我的实现中缺少了什么，有人可以帮助我解决这个问题？

Answer 1

问题在于配置usenativeencode的方式。它没有生效。此配置应起作用：

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

请注意usenativeencode是如何从原始配置中重新排列的。