我们如何为 Spring Cloud Stream Kafka 生产者、消费者和 KStreams 中的模式配置 value.subject.name.strategy？答案

【问题标题】：How can we configure value.subject.name.strategy for schemas in Spring Cloud Stream Kafka producers, consumers and KStreams?我们如何为 Spring Cloud Stream Kafka 生产者、消费者和 KStreams 中的模式配置 value.subject.name.strategy？
【发布时间】：2019-09-13 13:11:58
【问题描述】：

我想自定义 Spring Cloud Stream Producers、Consumers 和 KStreams 中 Avro 模式主题的命名策略。

这将在 Kafka 中使用 key.subject.name.strategy 和 value.subject.name.strategy 属性完成 -> https://docs.confluent.io/current/schema-registry/serializer-formatter.html#subject-name-strategy

在原生的 Kafka Producer 中，这是可行的：


private val producer: KafkaProducer<Int, Customer>

    init {
        val props = Properties()
        ...
        props[AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG] = "http://localhost:8081"
        props[AbstractKafkaAvroSerDeConfig.VALUE_SUBJECT_NAME_STRATEGY] = TopicRecordNameStrategy::class.java.name
        producer = KafkaProducer(props)
    }

    fun sendCustomerEvent(customer: Customer) {
        val record: ProducerRecord<Int, Customer> = ProducerRecord("customer", customer.id, customer)
        producer.send(record)
    }

但是我在 Spring Cloud Stream 中找不到如何执行此操作。到目前为止，我已经在制片人那里尝试过：

spring:
  application:
    name: spring-boot-customer-service
  cloud:
    stream:
      kafka:
        bindings:
          output:
            producer:
              configuration:
                key:
                  serializer: org.apache.kafka.common.serialization.IntegerSerializer
                value:
                  subject:
                    name:
                      strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

显然 Spring Cloud 使用自己的主题命名策略，接口为org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy，并且只有一个子类：DefaultSubjectNamingStrategy。

是否有配置value.subject.name.strategy 的声明方式，或者我们是否希望提供我们自己的org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy 实现和属性spring.cloud.stream.schema.avro.subject-naming-strategy？

【问题讨论】：

嗨，@codedependent，你知道了吗？我也有同样的担忧

标签： apache-kafka avro apache-kafka-streams spring-cloud-stream confluent-schema-registry

【解决方案1】：

正如另一个答案中指出的那样，有一个dedicated property、spring.cloud.stream.schema.avro.subjectNamingStrategy，允许为 Kafka 生产者设置不同的命名策略。

我贡献了org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy，它提供了开箱即用的功能。

对于 Kafka Streams 和本机序列化/反序列化（Spring Cloud Streams 3.0.0+ 的默认行为），您必须使用 Confluent 的实现 (io.confluent.kafka.serializers.subject.RecordNameStrategy) 和本机属性：

spring:
  application:
    name: shipping-service
  cloud:
    stream:
      ...
      kafka:
        streams:
          binder:
            configuration:
              application:
                id: shipping-service
              ...
              value:
                subject:
                  name:
                    strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy

【讨论】：

【解决方案2】：

你可以在你的属性中声明它为

spring.cloud.stream.schema.avro.subjectNamingStrategy=MyStrategy

其中 MyStrategy 是接口的实现。比如

object MyStrategy: SubjectNamingStrategy {
   override fun toSubject(schema: Schema): String = schema.fullName
}

【讨论】：