解决 Kafka Streams Avro 反序列化中的递归更新错误-java教程-PHP中文网

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

碧海醫心

发布： 2025-08-03 21:44:23

原创

241人浏览过

解决 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 中的字段名与类名冲突。以下是详细的步骤和建议：

检查 Avro Schema 文件：

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

在提供的例子中，PosInvoice.avsc 文件中的 DeliveryAddress 字段，其类型也是 DeliveryAddress，且字段名首字母大写，导致了混淆。
```
// 错误示例
{
  "type": "record",
  "name": "PosInvoice",
  "fields": [
    // ... other fields
    {"name": "DeliveryAddress", "type": "DeliveryAddress"}
  ]
}
```
登录后复制
修改字段命名：

将 Avro schema 中的字段名修改为小写，或者使用更具描述性的名称，以避免与类名冲突。
```
// 正确示例
{
  "type": "record",
  "name": "PosInvoice",
  "fields": [
    // ... other fields
    {"name": "deliveryAddress", "type": "DeliveryAddress"}
  ]
}
```
登录后复制
修改字段名后，需要重新生成 Avro 类。

使用 .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 类。

清理并重新编译:

在修改 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。