解决 Kafka Streams Avro 反序列化中的递归更新错误

本文档旨在帮助开发者解决在使用 Kafka Streams 和 Confluent Avro SerDe 时遇到的 java.lang.IllegalStateException: Recursive update 错误。该错误通常是由于 Avro schema 定义中的命名冲突导致的，我们将深入分析问题原因，并提供清晰的解决方案，以及最佳实践建议，确保你的 Kafka Streams 应用稳定可靠。

问题分析

java.lang.IllegalStateException: Recursive update 错误通常发生在 Kafka Streams 应用尝试反序列化 Avro 消息时。 根本原因是 Avro schema 中字段名与已存在的类名冲突，导致 Avro 试图递归地加载和初始化类，从而引发异常。

具体来说，当 Avro 反序列化器在 ConcurrentHashMap.computeIfAbsent 方法中尝试加载 schema 时，如果 schema 中的某个字段名与已存在的类名相同（忽略大小写），就会触发递归调用，最终导致 IllegalStateException。

解决方案

解决此问题的关键在于避免 Avro schema 中的字段名与类名冲突。以下是详细的步骤和建议：

1. 检查 Avro Schema 文件：

   仔细检查你的 .avsc 文件，特别是涉及到复杂类型（例如嵌套的 record 类型）的定义。确认字段名是否与任何已存在的类名冲突。

   在提供的例子中，PosInvoice.avsc 文件中的 DeliveryAddress 字段，其类型也是 DeliveryAddress，且字段名首字母大写，导致了混淆。

   // 错误示例
   {
     "type": "record",
     "name": "PosInvoice",
     "fields": [
       // ... other fields
       {"name": "DeliveryAddress", "type": "DeliveryAddress"}
     ]
   }

2. 修改字段命名：

   将 Avro schema 中的字段名修改为小写，或者使用更具描述性的名称，以避免与类名冲突。

   // 正确示例
   {
     "type": "record",
     "name": "PosInvoice",
     "fields": [
       // ... other fields
       {"name": "deliveryAddress", "type": "DeliveryAddress"}
     ]
   }

   修改字段名后，需要重新生成 Avro 类。

   

   企奶奶

   一款专注于企业信息查询的智能大模型，企奶奶查企业，像聊天一样简单。

   下载

3. 使用 .avdl 文件（推荐）：

   为了更好地管理 Avro schema 之间的依赖关系，建议使用 .avdl 文件来定义 Avro 类型。.avdl 文件允许你定义命名空间和 import 其他 schema，从而避免命名冲突。

   例如，你可以将 DeliveryAddress 定义在一个单独的 .avdl 文件中，并在 PosInvoice.avdl 中引用它。

   // DeliveryAddress.avdl
   @namespace("guru.learningjournal.kafka.examples.types")
   protocol DeliveryAddressProtocol {
     record DeliveryAddress {
       string addressLine;
       string city;
       string state;
       string pinCode;
     }
   }
   
   // PosInvoice.avdl
   @namespace("guru.learningjournal.kafka.examples.types")
   protocol PosInvoiceProtocol {
       import idl "DeliveryAddress.avdl";
   
       record PosInvoice {
           string invoiceNumber;
           string createdTime;
           string storeID;
           string posID;
           long customerID;
           string customerName;
           string email;
           string number;
           string paymentMethod;
           string deliveryType;
           DeliveryAddress deliveryAddress;
           string customerType;
           java.util.List lineItems;
           double totalAmount;
           double tax;
           double discount;
           double payableAmount;
       }
   
       record LineItem {
           string itemCode;
           String itemName;
           long itemQuantity;
           double itemPrice;
           double taxAmount;
           double totalValue;
       }
   }

   使用 Avro Maven 插件编译 .avdl 文件，生成对应的 Java 类。

4. 清理并重新编译:

   在修改 schema 后，确保清理 Maven 项目，并重新编译，以确保新的 Avro 类被正确生成和使用。

   mvn clean install

示例代码（更新后的 pom.xml）

    4.0.0

    guru.learningjournal.kafka.examples
    16-pos-fanout-avro
    1.0-SNAPSHOT

    
        11
        11
        11
    

    
        
            confluent
            https://packages.confluent.io/maven/
        
    

    
        
            org.apache.kafka
            kafka-streams
            3.3.1
        
        
            io.confluent
            kafka-streams-avro-serde
            7.2.2
        
        
            org.apache.logging.log4j
            log4j-slf4j-impl
            2.19.0
        
    

    
        
            
            
                org.apache.maven.plugins
                maven-compiler-plugin
                3.8.0
                
                    ${java.version}
                    ${java.version}
                
            

            
            
                org.apache.avro
                avro-maven-plugin
                1.11.0
                
                    
                        generate-sources
                        
                            idl
                        
                        
                            ${project.basedir}/src/main/resources/schema/
                            ${project.basedir}/src/main/java/
                            String
                            
                                ${project.basedir}/src/main/resources/schema/DeliveryAddress.avdl
                            
                        
                    
                
            
        
    

注意: 更新了 avro-maven-plugin 的版本到 1.11.0，并且将 goal 修改为 idl，以支持 .avdl 文件的编译。 确保在 configuration 节点中指定了 sourceDirectory、outputDirectory 和 stringType。

总结与最佳实践

解决 java.lang.IllegalStateException: Recursive update 错误的关键在于避免 Avro schema 中的命名冲突。 通过遵循以下最佳实践，可以有效地避免此类问题：

• 字段命名规范： 始终使用小写字母作为 Avro schema 字段名的开头。
• 使用 .avdl 文件： 使用 .avdl 文件来定义和管理 Avro schema，特别是当涉及到复杂的类型和依赖关系时。
• 清晰的命名空间： 为 Avro schema 定义清晰的命名空间，以避免与其他 schema 或类名冲突。
• 版本控制： 使用版本控制系统（如 Git）来管理 Avro schema 文件，以便追踪变更和回滚。
• 充分测试： 在生产环境中部署 Kafka Streams 应用之前，进行充分的测试，以确保 Avro 反序列化能够正常工作。

通过遵循这些建议，你可以有效地避免 java.lang.IllegalStateException: Recursive update 错误，并确保你的 Kafka Streams 应用稳定可靠。