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

Question

我想定制Spring云流生产者、消费者和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制作人中，这是这样的：

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:
  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.subject-naming-strategy实现和属性spring.cloud.stream.schema.avro.subject-naming-strategy？

Answer 1

正如另一个答案中指出的，有一个专用属性spring.cloud.stream.schema.avro.SubjectNamingStrategy，它允许为Kafka制作者设置不同的命名策略。

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

对于Kafka流和本机序列化/反序列化（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