AVRO-反序列化POJOs

Question

我是Avro和Kafka的新手，我花了几天时间来发送关于Kafka主题的序列化数据...不成功。

让我来解释一下我想要达到的目标：

在生产者方面，我通过SOAP接收数据并发送关于Kafka主题的内容。我正在使用CXF从WSDL生成POJO，并且编写了相应的模式。我正在尝试做的是序列化由CXF解封的对象，并在我的Kafka主题上发送它们。

在web上找到的大多数示例中，Avro记录都是使用已知的模式（或数据类型）生成的，但在这种情况下我不知道序列化数据时将使用哪种模式。因此我动态地获取消息类型（通过CXF拦截器）并以这种方式序列化：

// get unmarshaled POJO
MessageContentsList objs = MessageContentsList.getContentsList(message);
Object obj = objs.get(0);

EncoderFactory factory = EncoderFactory.get();
ByteArrayOutputStream out = new ByteArrayOutputStream();
Encoder encoder = factory.directBinaryEncoder(out, null);

// getting schema from class name (first approach)
String scName = obj.getClass().getSimpleName();
InputStream avroRes = this.getClass().getClassLoader().getResourceAsStream(scName);
Schema schema = new Schema.Parser().parse(avroRes);

ReflectDatumWriter<Object> writer = new ReflectDatumWriter<Object>(schema);
writer.write(obj, encoder);
encoder.flush();
out.close();

KeyedMessage< String, byte[]> kMessage = new KeyedMessage<String, byte[]>("mytopic", out.toByteArray());
producer.send(kMessage);

这样，我可以发送关于我的主题的数据，但我不能从传入的消息中获取模式。

有没有办法：

• 从Kafka主题中读取消息并获取用于序列化的架构？
• 在消耗和反序列化时将泛型记录映射到POJO？

当数据类型未知时，在Kafka主题上发送Avro记录的“最佳”实践是什么？

也许我在阅读Avro文档时错过了什么，没有按预期使用它。

谢谢你的帮助...

Answer 1

发送到Kafka主题的消息应该对模式和Avro记录进行编码。如果在每条消息中发送模式的开销太大，则改为发送该模式的标识符。消息使用者可以使用标识符从架构注册表检索完整的架构定义。例如，序列化Kafka消息的代码将模式标识符写入消息的前几个字节：

ByteArrayOutputStream out = new ByteArrayOutputStream();

schema = getSchema(object);
int id = schemaRegistry.register(subject, schema);
out.write(MAGIC_BYTE);
out.write(ByteBuffer.allocate(idSize).putInt(id).array());

BinaryEncoder encoder = encoderFactory.directBinaryEncoder(out, null);
DatumWriter<Object> writer;
if (object instanceof SpecificRecord) {
  writer = new SpecificDatumWriter<Object>(schema);
} else {
  writer = new GenericDatumWriter<Object>(schema);
}
writer.write(object, encoder);
encoder.flush();

byte[] bytes = out.toByteArray();
out.close();
return bytes;