消费者使用spring-cloud-stream-kafka-binder生成的spring-kafka生成的avro消息

Question

我试图使用ConsumerSeeKaware，阅读kafka主题中可用的最后一条消息。消息类型是Avro对象列表。我能成功地做到这一点。但在反序列化过程中会失败。该消息使用spring-cloud-stream-kafka框架生成。消息具有contentTypecontentType=application/x-java-object；type=java.util.ArrayList。

我知道avro消息可以像下面这样反序列化。

 DatumReader<GenericRecord> datumReader =
            new SpecificDatumReader<>(targetType.newInstance().getSchema());
        Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);

        result = (T) datumReader.read(null, decoder);

但不管用。可能是因为两件事。

>

消息是avro对象的列表。但我正在尝试使用Avro模式创建DataReader。但是我尝试创建schema.createArray（userDTo.class）这样的模式。但它并不起作用。

我认为avro消息的预期内容类型是application/avro，但当消息由s-c-s生成时，它是contenttype=application/x-java-object；

我试图通过实现org.apache.kafka.common.serialization.deserializer来创建一个反序列化程序，并构造KafkaConsumerFactory。有人能帮忙吗？

Answer 1

请参阅property...Producer.UsenativeEncode（如生成器属性中所示）。

使用编码

当设置为true时，出站消息由客户端库直接序列化，客户端库必须进行相应的配置（例如，设置适当的Kafka生产者值序列化器）。使用此配置时，出站消息封送不基于绑定的contentType。当使用本机编码时，使用者有责任使用适当的解码器（例如：Kafka使用者值反序列化器）来反序列化入站消息。此外，当使用本机编码/解码时，headerMode属性将被忽略，头将不会嵌入到消息中。