Java中Avro Schema向后兼容异常排查求助

问题背景 #

尝试为Avro Schema添加新字段并设置默认值，期望实现向后兼容，但使用新Schema（v2_0.avsc）消费旧Schema（v1_0.avsc）写入的消息时，始终抛出java.io.EOFException。已尝试多种联合类型与默认值的组合，Avro版本1.8.1和1.11.0均出现相同问题。

旧Schema（v1_0.avsc） #

{
  "type": "record",
  "name": "Test",
  "fields": [
     {"type": "string",  "name": "field_1"}
    ,{"type": "string",  "name": "field_2"}
    ]
}

新Schema（v2_0.avsc） #

{
  "type": "record",
  "name": "Test",
  "fields": [
     {"type": "string",  "name": "field_1"}
    ,{"type": "string",  "name": "field_2"}
    ,{"type": "string",  "name": "field_3", "default": "default_value"}
    ]
}

尝试过的新字段组合 #

• 带默认字符串的联合类型（顺序1）：

  ,{"type": ["string", "null"], "name": "field_3", "default": "default_value"}
  

• 带默认null的联合类型（顺序2）：

  ,{"type": ["null", "string"], "name": "field_3", "default": "null"}
  

• 仅设置null默认值：

  ,{"type": ["null", "string"], "name": "field_3", "default": null}
  

抛出的异常栈 #

java.io.EOFException: null
    at org.apache.avro.io.BinaryDecoder.ensureBounds(BinaryDecoder.java:542)
    at org.apache.avro.io.BinaryDecoder.readInt(BinaryDecoder.java:173)
    at org.apache.avro.io.BinaryDecoder.readIndex(BinaryDecoder.java:493)
    at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:282)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:188)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
    at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:260)
    at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:142)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:248)
    at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:154)

消费者核心代码 #

byte[] data = record.value();
// 原错误写法：仅传入新Schema
DatumReader<GenericRecord> reader = new SpecificDatumReader<>(SCHEMA);
Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);
GenericRecord message;
message = reader.read(null, decoder);

生产者核心代码 #

GenericRecord payload = new GenericData.Record(SCHEMA);
payload.put("field_1", "001");
payload.put("field_2", "002");
DatumWriter<GenericRecord> writer = new SpecificDatumWriter<>(SCHEMA);
byte[] serializedBytes;
try ( ByteArrayOutputStream out = new ByteArrayOutputStream() ) {
    BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
    writer.write(payload, encoder);
    encoder.flush();
    serializedBytes = out.toByteArray();
}
ProducerRecord<String, byte[]> message = new ProducerRecord<>(TOPIC, serializedBytes);
producer.send(message);

解决方案 #

核心问题 #

创建SpecificDatumReader时仅传入了读取用的新Schema，未指定写入时的旧Schema。Avro的Schema Resolution机制需要明确知道数据是用哪个Schema写入的，才能正确识别缺失的字段并填充默认值；若只传一个Schema，Avro会默认写入和读取Schema完全一致，导致解析旧消息时试图读取不存在的字段数据，触发EOFException。

修改代码 #

将消费者中创建DatumReader的代码改为：

// 先加载旧版本的v1_0.avsc为OLD_SCHEMA对象
Schema OLD_SCHEMA = new Schema.Parser().parse(new File("path/to/v1_0.avsc"));
// 传入写入Schema（旧）和读取Schema（新）
DatumReader<GenericRecord> reader = new SpecificDatumReader<>(OLD_SCHEMA, SCHEMA);

补充说明 #

• 若使用GenericDatumReader，同样需要传入两个Schema，原理完全一致；
• 若使用Schema Registry管理Schema，可通过Registry获取对应版本的写入Schema，无需本地维护多版本文件。

内容的提问来源于stack exchange，提问作者Pedro