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

Question

我使用kafka流来消耗来自一个主题的JSON字符串，处理并生成存储在另一个主题中的响应。然而，需要对响应主题产生的消息需要采用avro格式。

我已经尝试使用键作为字符串序列和值作为规范AvroSerde

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

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。您应该在使用流时使用正确的SERDE，在发布流时使用相同的SERDE。

如果您的输入是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());