Apache Beam Schema更新兼容性测试：如何验证V2可解码V1类实例

兼容性修正说明 #

你当前的TransactionPojoV2定义默认不兼容旧数据：新增的fee是必填字段，旧序列化数据中没有对应值，解码时会直接抛出缺失字段异常。你需要给新增字段指定默认值，调整后的V2代码如下：

@DefaultSchema(JavaFieldSchema.class)
public class TransactionPojoV2 {
  public final String bank;
  public final double purchaseAmount;
  public final double fee;

  @SchemaCreate
  public TransactionPojoV2(
      String bank, 
      double purchaseAmount, 
      // 给新增字段指定默认值，旧数据解码时自动填充该值
      @DefaultValue.DoubleDefault(0.0) double fee) {
    this.bank = bank;
    this.purchaseAmount = purchaseAmount;
    this.fee = fee;
  }
}

如果新增字段允许为空，也可以将fee改为Double包装类，给构造参数加@Nullable注解，旧数据解码后fee会自动填充为null。

兼容性测试实现 #

测试需使用Dataflow管道实际采用的SchemaCoder进行编解码，才能真实模拟线上序列化逻辑，示例测试代码如下：

import org.apache.beam.sdk.schemas.SchemaCoder;
import org.apache.beam.sdk.schemas.utils.JavaFieldSchema;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.junit.Test;
import java.io.ByteArrayInputStream;
import static org.junit.Assert.assertEquals;

public class SchemaCompatibilityTest {

  @Test
  public void testV1ToV2DecodeCompatibility() throws Exception {
    // 1. 分别获取V1、V2类对应的SchemaCoder
    SchemaCoder<TransactionPojoV1> v1Coder = SchemaCoder.of(
        new JavaFieldSchema().schemaFor(TransactionPojoV1.class),
        TransactionPojoV1.class,
        new JavaFieldSchema()
    );
    SchemaCoder<TransactionPojoV2> v2Coder = SchemaCoder.of(
        new JavaFieldSchema().schemaFor(TransactionPojoV2.class),
        TransactionPojoV2.class,
        new JavaFieldSchema()
    );

    // 2. 序列化V1实例
    TransactionPojoV1 v1Obj = new TransactionPojoV1("招商银行", 299.9);
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    v1Coder.encode(v1Obj, outputStream);
    byte[] serializedData = outputStream.toByteArray();

    // 3. 用V2的Coder反序列化V1的字节数据
    ByteArrayInputStream inputStream = new ByteArrayInputStream(serializedData);
    TransactionPojoV2 v2Obj = v2Coder.decode(inputStream);

    // 4. 断言字段符合预期
    assertEquals(v1Obj.bank, v2Obj.bank);
    assertEquals(v1Obj.purchaseAmount, v2Obj.purchaseAmount, 0.001);
    // 校验新增字段的默认值符合预期
    assertEquals(0.0, v2Obj.fee, 0.001);
  }
}

无停服更新注意事项 #

• 必须先升级所有下游消费节点到V2版本，再升级上游生产节点写入V2数据，避免旧版本节点无法识别V2的新增字段
• 如果后续需要删除字段，不要直接删除字段定义，先给要删除的字段加@SchemaIgnore注解，等全链路都升级后再完全删除字段

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