Kafka流：序列化回avro

Question

我试图构建一个流，它获得一个Avro主题，做一个简单的转换，然后以Avro格式再次将其发送回另一个主题，我有点卡在最后的序列化部分。

我创建了一个AVRO模式，我正在导入它并使用它创建特定的AVRO Serde。但是我不知道如何使用这个serde将电影对象序列化回AVRO。

这是流类：

class StreamsProcessor(val brokers: String, val schemaRegistryUrl: String) {

    private val logger = LogManager.getLogger(javaClass)

    fun process() {
        val streamsBuilder = StreamsBuilder()

        val avroSerde = GenericAvroSerde().apply {
            configure(mapOf(Pair("schema.registry.url", schemaRegistryUrl)), false)
        }

        val movieAvro = SpecificAvroSerde<Movie>().apply{
            configure(mapOf(Pair("schema.registry.url", schemaRegistryUrl)), false)
        }

        val movieAvroStream: KStream<String, GenericRecord> = streamsBuilder
                .stream(movieAvroTopic, Consumed.with(Serdes.String(), avroSerde))

        val movieStream: KStream<String, StreamMovie> = movieAvroStream.map {_, movieAvro ->
            val movie = StreamMovie(
                    movieId = movieAvro["name"].toString() + movieAvro["year"].toString(),
                    director = movieAvro["director"].toString(),
            )
             KeyValue("${movie.movieId}", movie)
        }

        // This where I'm stuck, the call is wrong because movieStream is not a <String, movieAvro> object 
        movieStream.to(movieTopic, Produced.with(Serdes.String(), movieAvro))

        val topology = streamsBuilder.build()

        val props = Properties()
        props["bootstrap.servers"] = brokers
        props["application.id"] = "movies-stream"
        val streams = KafkaStreams(topology, props)
        streams.start()
    }
}

谢谢

Answer 1

结果流的类型是KStream

为什么要尝试使用specifiavroserde

比较https://github.com/confluentinc/kafka-streams-examples/blob/5.4.1-post/src/test/java/io/confluent/examples/streams/SpecificAvroIntegrationTest.java