在KStream中聚合Avro对象ArrayList时遭遇类型转换与序列化异常

异常根源 #

1. 类型转换异常：Kafka Streams默认基于Avro的SpecificRecord类型做序列化/反序列化，但你直接使用了Java原生ArrayList，框架会尝试将ArrayList强制转为SpecificRecord，导致类型不匹配。
2. 序列化异常：聚合结果是ArrayList<Employee>，没有对应的Avro序列化器支撑，无法将其序列化为Avro消息写入repartition主题或状态存储。

解决方案 #

1. 定义Avro集合类型 #

不能直接用Java原生集合，需要创建Avro Schema来封装Employee列表，比如创建EmployeeList.avsc文件：

{
  "type": "record",
  "name": "EmployeeList",
  "namespace": "你的业务包路径",
  "fields": [
    {"name": "employees", "type": {"type": "array", "items": "你的业务包路径.Employee"}}
  ]
}

通过Avro工具生成对应的EmployeeList类（该类会继承SpecificRecord，符合Avro序列化要求）。

2. 修改聚合代码逻辑 #

将聚合结果类型替换为EmployeeList，调整初始化、添加、移除逻辑，同时显式指定状态存储的序列化器：

final KTable<String, EmployeeList> caKTables = caKTable
    .groupBy((key, value) -> pair(value.getBrStId(), value))
    .aggregate(
        // 初始化空的EmployeeList容器
        () -> {
            EmployeeList empList = new EmployeeList();
            empList.setEmployees(new ArrayList<>());
            return empList;
        },
        // 新增Employee到聚合列表
        (key, value, aggregate) -> {
            aggregate.getEmployees().add(value);
            return aggregate;
        },
        // 移除旧Employee记录
        (key, oldValue, agg) -> {
            agg.getEmployees().remove(oldValue);
            return agg;
        },
        // 指定状态存储的序列化配置
        Materialized.<String, EmployeeList, KeyValueStore<Bytes, byte[]>>as("employee-aggregate-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(AvroSerdes.specific(EmployeeList.class))
    );

3. 确认全局序列化配置 #

确保Kafka Streams全局配置中，默认值序列化器为Avro的SpecificAvroSerde，并配置Schema Registry地址：

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsProps.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class.getName());
streamsProps.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://你的schema-registry地址:8081");

4. 适配Join逻辑 #

修改caJoiner的实现，适配EmployeeList类型，比如通过empList.getEmployees()获取列表数据进行后续业务处理。

关键注意点 #

• 必须使用Avro定义的集合类型替代Java原生集合，因为Avro序列化器仅支持SpecificRecord/GenericRecord类型的序列化。
• 聚合时显式指定序列化器非常重要，避免全局默认序列化器与聚合结果类型不匹配引发异常。
• 确保Employee类正确实现equals()和hashCode()方法，否则remove(oldValue)无法精准定位并移除目标元素。

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