Spring Cloud Streams kafka活页夹-主题序列化配置

Question

因此，我认为我自己陷入了困惑，因为我知道SpringCloudStreams有两种不同的Kafka活页夹：

• 《春云流》Kafka活页夹

我正在寻找正确的YAML设置，以便在spring cloud streams的常规kafka活页夹中定义序列化器和反序列化器：

我可以使用以下逻辑调整默认值：

spring:
  main:
    web-application-type: NONE
  application:
    name: tbfm-translator
  kafka:
    consumer:
      group-id: ${consumer_id}
    bootstrap-servers: ${kafka_servers}
  cloud:
    schemaRegistryClient:
      endpoint: ${schema_registry}
    stream:
#      default:
#        producer.useNativeEncoding: true
#        consumer.useNativeEncoding: true
      defaultBinder: kafka
      kafka:
        binder:
          auto-add-partitions: true # I wonder if its cause this is set
          auto-create-topics: true # Disabling this seem to override the server setings and will auto create

          producer-properties:
            # For additional properties you can check here:
            # https://docs.confluent.io/current/installation/configuration/producer-configs.html

            schema.registry.url: ${schema_registry}

            # Disable for auto schema registration
            auto.register.schemas: false

            # Use only the latest schema version
            use.latest.version: true

            # This will use reflection to generate schemas from classes - used to validate current data set
            # against the scheam registry for valid production
            schema.reflection: true

            # To use an avro key enable the following line
            #key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

            #This will use a string based key - aka not in the registry - dont need a name strategy with string serializer
            key.serializer: org.apache.kafka.common.serialization.StringSerializer

            # This will control the Serializer Setup
            value.subject.name.strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

即：

spring.cloud.stream.kafka.binder.producer-properties.value.serializer
spring.cloud.stream.kafka.binder.producer-properties.key.serializer

我想我应该能够在每个主题的基础上做到这一点：

spring:
  cloud:
    stream:
      bindings:
        my-topic:
          destination: a-topic
          xxxxxxxx??

我遇到了设置：

          producer:
            use-native-encoding: false
            keySerde: <CLASS>

但这似乎不起作用。我可以设置一个简单的属性来在每个主题的基础上执行此操作吗？我认为keySerde用于Kafka-stream实现，而不是普通的kafka绑定器。

Answer 1

    stream:
      bindings: # Define output topics here and then again in the kafka.bindings section
        test:
          destination: multi-output
          producer:
            useNativeDecoding: true

      kafka:
        bindings:
          test:
            destination: multi-output
            producer:
              configuration:
                value.serializer: org.apache.kafka.common.serialization.StringSerializer

这似乎工作-但很烦人我必须在两个地方复制绑定定义

让我们想回避YAML风格的定义

Answer 2

使用本机编码必须为true才能使用您自己的序列化程序。

spring.cloud.stream.kafka.bindings.my-topic.producer.configuration.value.serializer：...

有关Kafka特定的制作人属性，请参阅文档。

配置

使用包含通用Kafka生产者属性的键/值对进行映射。

默认值：空地图。