java - 如何从 Java 消费者的主题中的消息中从模式注册表中检索 AVRO 模式

Question

我们正在使用带有模式注册表的 Confluent Kafka。我们有 40 多个主题。我们的应用程序使用注册表中的模式将 avro 消息写入这些主题。

据我了解，在使用注册表时，该消息不包含实际的架构，而是对注册表中的架构（架构 id）的引用。我正在开发一个 Java 实用程序，它将接受主题名称或主题名称列表，并从每个主题中检索有限数量（可能 50 条）的消息。然后，此实用程序应将 avro 消息转换为 json。

我见过的所有示例，仍然必须提供能够将 byte[] 转换为 json 的模式。

我希望并想知道，是否可以使用来自主题的 avro 消息（模式 id）中的信息动态地从注册表中获取模式，并使用该信息将消息转换为 json。

有可能这样做吗？有人可以给我一个关于如何实现这一目标的例子吗？

谢谢

更新

@eik

试验 1

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    "io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer");
final Consumer<String, GenericRecord> genericConsumer = new KafkaConsumer<>(props);
genericConsumer.subscribe(Collections.singletonList("TOPICNAME"));            
while (true) {
    final ConsumerRecords<String, GenericRecord> genericConsumerRecords = genericConsumer.poll(Duration.ofMillis(1000));
    System.out.println("genericConsumerRecords.count() : " + genericConsumerRecords.count() + " genericConsumerRecords.isEmpty() : " + genericConsumerRecords.isEmpty());
    genericConsumerRecords.forEach(genericRecord1 -> {
        try {
            System.out.println("convert(genericRecord1.value()) -> " + convert(genericRecord1.value()));
        } catch (IOException e) {
            e.printStackTrace();
        }
        //
    });
}

这是输出 genericConsumerRecords.count() : 0 genericConsumerRecords.isEmpty() : true

注意：：转换方法是下面答案中的方法

试验 2

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
   "org.apache.kafka.common.serialization.ByteArrayDeserializer");
final Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("TOPICNAME"));

while (true) {
    final ConsumerRecords<String, byte[]> consumerRecords = consumer.poll(Duration.ofMillis(1000));
    System.out.println("consumerRecords.count() : " + consumerRecords.count() + " consumerRecords.isEmpty() : " + consumerRecords.isEmpty());
    consumerRecords.forEach(record1 -> {
       String string = new String(record1.value(), StandardCharsets.UTF_8);
       System.out.printf("offset = %d, key = %s, value = %s \n", record1.offset(), record1.key(), string);
    });
}

这是输出

consumerRecords.count() : 60 consumerRecords.isEmpty() : false
offset = 0, key = e3bff195-08a7-4c58-99de-98ffe2d460e6, value = He52d6fa6-841f-430c-8bf7-bd4c7b684129 http://schemaregistryurl:8081/subjects/TOPICNAME-value/versions/1/schema Canon Message to represent CustomerPrefAVRFAST 162019-08-07T08:35:35.9950728 QA1-Test-0421-16$CustomerPrefData 1He52d6fa6-841f-430c-8bf7-bd4c7b684129 RawH862437d0-e260-45f9-ab5e-345b536d685a02020-04-21T17:48:52.601Z$CustomerPref POL_MAST02020-04-21T11:17:28.241ZHe3bff195-08a7-4c58-99de-98ffe2d460e69
False&1900-01-01T00:00:00He3bff195-08a7-4c58-99de-98ffe2d460e6He3bff195-08a7-4c58-99de-98ffe2d460e6 

注意：：我必须从输出中删除一些非 ascii 字符

第二种方法确实给出了输出，但它是一个byte[]，需要得到json输出。尝试了不同的方法都没有成功。

我如何解决它？

谢谢

标签： javaapache-kafkakafka-consumer-apiavroconfluent-schema-registry

Answer 1

一种选择是使用io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer获取标准的 Apache Avro GenericRecord，然后执行以下操作：

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;

class GenericRecord2Json {
  String convert(GenericRecord record) throws IOException {
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

    Schema schema = record.getSchema();
    DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(schema);
    Encoder encoder = EncoderFactory.get().jsonEncoder(schema, outputStream, /* pretty= */ true);
    writer.write(record, encoder);
    encoder.flush();

    return outputStream.toString(StandardCharsets.UTF_8);
  }
}

它以 JSON 格式返回您的消息。Schema 由 检索record.getSchema()。