字段顺序存在Avro模式Java deepcopy问题

Question

我目前正在寻找意外行为的解决方案，这些意外行为是在使用Java处理特定的AVRO模式演化场景时出现的，我还在消费者中做了一次深度复制，将GenericRecord类解析为从AVRO模式生成的特定类。

为了解释正在发生的事情，我将使用一个简化的模式示例：

{
  "name":"SimpleEvent",
  "type":"record",
  "namespace":"com.simple.schemas",
  "fields":[
     {
        "name":"firstfield",
        "type":"string",
        "default":""
     },
     {
        "name":"secondfield",
        "type":"string",
        "default":""
     },
     {
        "name":"thirdfield",
        "type":"string",
        "default":""
     }
  ]
}

这只是一个简单的模式，有三个字符串字段，都是可选的，因为它们都有默认值。假设在某个时候我想添加另一个字符串字段，并删除一个字段，因为不再需要它，您最终会得到这样的结果：

{
  "name":"SimpleEvent",
  "type":"record",
  "namespace":"com.simple.schemas",
  "fields":[
     {
        "name":"firstfield",
        "type":"string",
        "default":""
     },
     {
        "name":"secondfield",
        "type":"string",
        "default":""
     },
     {
        "name":"newfield",
        "type":"string",
        "default":""
     }
  ]
}

这不应该破坏根据模式演化规则所做的更改。然而，当生产者开始使用较新的模式生成事件时，下游消费者中会发生一些奇怪的事情。

事实证明，生成的Java类（我使用Gradle avro插件生成类，但maven插件和avro工具命令行代码生成产生相同的输出）只查看字段顺序，而不是根据它们的名称映射字段。

这意味着字段“newfield”的值由使用旧版本的架构读取数据的下游使用者映射到“第三字段”。

我发现了一些基于名称执行手动映射的工作，但是，这对嵌套对象不起作用。

通过一些局部实验，我还发现了另一种可以正确解决架构差异的方法：

    Schema readerSchema = SimpleEvent.getClassSchema();
    Schema writerSchema = request.getSchema();

    if (readerSchema.equals(writerSchema)){
        return (SimpleEvent)SpecificData.get().deepCopy(writerSchema, request);
    }

    DatumWriter<GenericRecord> writer = new SpecificDatumWriter<>(writerSchema);
    BinaryEncoder encoder = null;
    ByteArrayOutputStream stream = new ByteArrayOutputStream();
    encoder = EncoderFactory.get().binaryEncoder(stream, encoder);

    writer.write(request, encoder);
    encoder.flush();

    byte[] recordBytes = stream.toByteArray();

    Decoder decoder = DecoderFactory.get().binaryDecoder(recordBytes, null);

    SpecificDatumReader<SimpleEvent> specificDatumReader = new SpecificDatumReader(writerSchema, readerSchema);
    SimpleEvent result = specificDatumReader.read(null, decoder);
    return result;

但是，这似乎是一种相当浪费/不优雅的方法，因为您首先必须将GenericRecord转换为byteArray，然后使用IcitaryDatumReader再次读取它。

deepcopy和datumreader类之间的区别在于，datumReader类似乎可以适应编写器模式不同于读取器模式的情况。

我觉得应该/可以有一个更好、更优雅的方法来处理这个问题。我真的很感激任何帮助/提示。

感谢提前：）

奥斯卡

Answer 1

在进一步挖掘和研究了我们之前在侦听器中使用的Kafkavrodeserializer之后，我注意到AbstractKafkarvodeserialiser有一个反序列化函数，可以在读取器模式中传递。这看起来不错，但它的工作！

package com.oskar.generic.consumer.demo;

import com.simple.schemas;

import io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

public class SimpleEventDeserializer extends AbstractKafkaAvroDeserializer implements Deserializer<Object> {

private boolean isKey;

@Override
public void configure(Map<String, ?> configs, boolean isKey) {
    this.isKey = isKey;
    configure(new KafkaAvroDeserializerConfig(configs));
}

@Override
public Object deserialize(String s, byte[] bytes) {
    return super.deserialize(bytes, SimpleEvent.getClassSchema());
}

@Override
public void close() {

}
}

然后在消费工厂中使用，如下所示：

@Bean
public ConsumerFactory<String, GenericRecord> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29095");
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "one");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SimpleEventDeserializer.class);

    return new DefaultKafkaConsumerFactory<>(props);
}

侦听器代码本身如下所示：

 @KafkaListener(topics = "my-topic")
public GenericRecord listen(@Payload GenericRecord request, @Headers MessageHeaders headers) throws IOException {
    SimpleEvent event = (SimpleEvent) SpecificData.get().deepCopy(request.getSchema(), request);
    return request;
}