Kafka Streams 可以使用一种格式的消息并生成另一种格式，例如 AVRO 消息

Question

我正在使用 kafka 流来使用来自一个主题的 JSON 字符串，处理并生成响应以存储在另一个主题中。但是，需要为响应主题生成的消息需要采用 avro 格式。

我试过使用键作为字符串 serde 和值作为 SpecificAvroSerde

以下是我创建拓扑的代码：

StreamsBuilder builder = new StreamsBuilder();
KStream<Object, Object> consumerStream =builder.stream(kafkaConfiguration.getConsumerTopic());
consumerStream = consumerStream.map(getKeyValueMapper(keyValueMapperClassName));
consumerStream.to(kafkaConfiguration.getProducerTopic());

以下是我的配置

    if (schemaRegistry != null && schemaRegistry.length > 0) {
        streamsConfig.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, String.join(",", schemaRegistry));          
    }
    streamsConfig.put(this.keySerializerKeyName, StringSerde.class);
    streamsConfig.put(this.valueSerialzerKeyName, SpecificAvroSerde.class);
    streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
    streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
    streamsConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, batchSize);
    streamsConfig.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, FailOnInvalidTimestamp.class);
    streamsConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
    streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Integer.parseInt(commitIntervalMs));
    streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numberOfThreads);
    streamsConfig.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, replicationFactor);
    streamsConfig.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
    streamsConfig.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
    streamsConfig.put(StreamsConfig.TOPOLOGY_OPTIMIZATION,StreamsConfig.OPTIMIZE);
    streamsConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionMode);
    streamsConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);

我在尝试示例时看到以下错误：

org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

Answer 1

问题出在键值 Serdes 上。您应该在使用流和发布流时使用正确的 serdes。

如果您的输入是 JSON 并且您想发布为 Avro，您可以按以下方式进行：

Properties streamsConfig= new Properties();
streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
StreamsBuilder builder = new StreamsBuilder();
KStream<Object, Object> consumerStream =builder.stream(kafkaConfiguration.getConsumerTopic(),Consumed.with(Serdes.String(), Serdes.String()));

// Replace AvroObjectClass with your avro object type
KStream<String,AvroObjectClass> consumerAvroStream = consumerStream.map(getKeyValueMapper(keyValueMapperClassName));
consumerAvroStream.to(kafkaConfiguration.getProducerTopic());

Kafka Streams 可以使用一种格式的消息并生成另一种格式，例如 AVRO 消息

Can Kafka Streams consume message in a format and produce another format such as AVRO message

apache-kafka-streams