KafkaStreamsAvro反序列化错误解决方法

在使用 Kafka Streams 和 Confluent Avro SerDe 时，你是否遇到过 `java.lang.IllegalStateException: Recursive update` 错误？本文将深入剖析这一常见问题的根源，该错误通常源于 Avro schema 定义中存在的命名冲突。我们将详细分析 Avro 反序列化过程中字段名与已存在类名冲突导致递归调用的机制，并提供清晰、可操作的解决方案，包括如何检查和修改 Avro schema 文件、利用 `.avdl` 文件管理 schema 依赖关系、以及如何使用 Avro Maven 插件进行编译。此外，本文还提供了更新后的 `pom.xml` 示例代码，助你快速解决问题。遵循本文的最佳实践建议，确保你的 Kafka Streams 应用稳定可靠，避免此类异常的发生。

本文档旨在帮助开发者解决在使用 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<LineItem> 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）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>guru.learningjournal.kafka.examples</groupId>
    <artifactId>16-pos-fanout-avro</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <java.version>11</java.version>
    </properties>

    <repositories>
        <repository>
            <id>confluent</id>
            <url>https://packages.confluent.io/maven/</url>
        </repository>
    </repositories>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-streams-avro-serde</artifactId>
            <version>7.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>2.19.0</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- Maven Compiler Plugin-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>

            <!-- Maven Avro plugin for generating pojo-->
            <plugin>
                <groupId>org.apache.avro</groupId>
                <artifactId>avro-maven-plugin</artifactId>
                <version>1.11.0</version>
                <executions>
                    <execution>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>idl</goal>
                        </goals>
                        <configuration>
                            <sourceDirectory>${project.basedir}/src/main/resources/schema/</sourceDirectory>
                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
                            <stringType>String</stringType>
                            <imports>
                                <import>${project.basedir}/src/main/resources/schema/DeliveryAddress.avdl</import>
                            </imports>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

注意: 更新了 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 应用稳定可靠。

本篇关于《KafkaStreamsAvro反序列化错误解决方法》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！