使用avro模式注册表的java kafka stream的正确指南

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.10.2</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>6.2.0</version>
</dependency>

String schemaRegistryUrl = "http://localhost:8081";
SchemaRegistryClient schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 1000);

String avroSchema = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"}]}";
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(avroSchema);

int schemaId = schemaRegistryClient.register("my-topic-value", schema);

Properties props = new Properties();
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

KStream<String, GenericRecord> inputStream = builder.stream("my-topic", Consumed.with(Serdes.String(), SpecificAvroSerde.<GenericRecord>as()));
KStream<String, GenericRecord> outputStream = inputStream.mapValues(value -> {
    // 处理逻辑
    return transformedValue;
});
outputStream.to("output-topic");