Spring-cloud kafka流模式注册表

Question

我试图用函数编程（和spring cloud stream）转换来自输入主题的输入AVRO消息，并在输出主题上发布新消息。下面是我的转换函数：

@Bean
public Function<KStream<String, Data>, KStream<String, Double>> evenNumberSquareProcessor() {
    return kStream -> kStream.transform(() -> new CustomProcessor(STORE_NAME), STORE_NAME);
}

    spring:
  cloud:
    stream:
      function:
        definition: evenNumberSquareProcessor
      bindings:
        evenNumberSquareProcessor-in-0:
          destination: input
          content-type: application/*+avro
          group: group-1
        evenNumberSquareProcessor-out-0:
          destination: output
      kafka:
        binder:
          brokers: my-cluster-kafka-bootstrap.kafka:9092
          consumer-properties:
            value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
            schema.registry.url: http://localhost:8081

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: my-cluster-kafka-bootstrap.kafka:9092
            configuration:
              schema.registry.url: http://localhost:8081
              default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          bindings:
            evenNumberSquareProcessor-in-0:
              consumer:
                destination: input
                valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
            evenNumberSquareProcessor-out-0:
                destination: output

我的spring boot应用程序是以这种方式声明的，并激活了模式注册表客户机：

    @EnableSchemaRegistryClient
@SpringBootApplication
public class TransformApplication {
    public static void main(String[] args) {
        SpringApplication.run(TransformApplication.class, args);
    }
}

谢谢你能给我带来的任何帮助。

视CG

Answer 1

在配置下配置架构注册表，然后所有绑定器都可以使用它。对了。avro序列化程序位于绑定和特定通道下。如果要使用默认属性default.value.serde:。你的Serde可能也错了。

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: localhost:9092
            configuration:
              schema.registry.url: http://localhost:8081
              default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          bindings:
            process-in-0:
              consumer:
                valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

不要使用@enableschemareGistryClient。在Avro Serde上启用架构注册表。在本例中，我使用定义的beandata。试着在这里遵循这个例子。

@Service
public class CustomSerdes extends Serdes {

    private final static Map<String, String> serdeConfig = Stream.of(
            new AbstractMap.SimpleEntry<>(SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"))
            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

    public static Serde<Data> DataAvro() {
        final Serde<Data> dataAvroSerde = new SpecificAvroSerde<>();
        dataAvroSerde.configure(serdeConfig, false);
        return dataAvroSerde;
    }
}