基于Kafka2的IgniteSinkConnector在Kafka3集群运行异常排查

问题描述 #

在部署了Strimzi的Kubernetes集群中，搭配Kafka 3运行IgniteSinkConnector，已添加全部依赖且连接器已加载，但出现IgniteSinkTask中的静态类无法初始化的异常（Jar包确认包含该类）。该连接器基于Kafka 2开发，需解决两个核心问题：

1. Kafka 3环境中是否可以使用基于Kafka 2开发的SinkConnector？
2. 如何修复该静态类初始化异常？

错误日志 #

2023-07-11 23:41:44,140 ERROR [ignite-sink-connector|task-1] WorkerSinkTask{id=ignite-sink-connector-1} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. Error: Could not initialize class org.apache.ignite.stream.kafka.connect.IgniteSinkTask$StreamerContext$Holder (org.apache.kafka.connect.runtime.WorkerSinkTask) [task-thread-ignite-sink-connector-1]
java.lang.NoClassDefFoundError: Could not initialize class org.apache.ignite.stream.kafka.connect.IgniteSinkTask$StreamerContext$Holder
    at org.apache.ignite.stream.kafka.connect.IgniteSinkTask$StreamerContext.getStreamer(IgniteSinkTask.java:198)
    at org.apache.ignite.stream.kafka.connect.IgniteSinkTask.put(IgniteSinkTask.java:118)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:583)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
    at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:177)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.base/java.lang.Thread.run(Thread.java:833)

解决方案 #

一、Kafka 3与Kafka 2 SinkConnector的兼容性 #

Kafka 3和Kafka 2的Connect API整体具备有限兼容性：

• 若连接器仅使用了Connect API的公开标准接口（如SinkConnector、SinkTask的核心方法），大概率可以直接运行；
• 但如果连接器依赖了Kafka内部未公开的API、特定版本的类结构，或者依赖的kafka-clients、connect-api版本与Kafka 3差异较大，就会出现类加载、初始化异常。

简言之：不是所有Kafka 2的连接器都能无缝在Kafka 3下运行，需要结合依赖和API使用情况验证适配性。

二、静态类初始化异常的修复步骤 #

注意：java.lang.NoClassDefFoundError: Could not initialize class本质不是类找不到，而是类的静态代码块或静态成员初始化失败，JVM因此标记该类为不可用。针对这个问题，按以下步骤排查修复：

1. 解决Kafka依赖版本冲突

   • 问题根源：IgniteSinkConnector基于Kafka 2构建，其依赖的kafka-clients、connect-api版本与Kafka 3环境的版本不一致，导致类方法签名、内部结构不匹配，触发初始化失败。
   • 修复方式：
     • 重新编译连接器，将依赖的Kafka相关jar包替换为与集群一致的Kafka 3.x版本；
     • 若无法自行编译，在连接器的依赖中排除Kafka相关包（如kafka-clients、connect-api），让Strimzi集群的类加载器提供这些类。

2. 调整Strimzi的类加载隔离策略

   • 问题根源：Strimzi默认启用类加载隔离（isolated模式），连接器的类加载器与集群的Kafka类加载器相互隔离，导致静态类初始化时无法访问必要的依赖类。
   • 修复方式：
     • 在Strimzi的KafkaConnect资源配置中，设置config.classLoaderMode: shared，让连接器共享集群的Kafka类加载器；
     • 或者将连接器的所有依赖jar包放到Strimzi Connect集群的共享插件目录（对应持久化卷的plugins路径）。

3. 验证Ignite依赖与Kafka 3的兼容性

   • 问题根源：旧版Ignite可能与Kafka 3使用的Java版本（如Java 11+）、依赖库存在冲突，导致静态初始化环节出错。
   • 修复方式：
     • 确认Ignite版本支持当前集群使用的Java版本；
     • 将连接器依赖的Ignite相关jar包更新到与Kafka 3兼容的版本，重新打包后部署。

4. 排查初始化失败的具体根源

   • 问题根源：当前日志仅显示初始化失败的结果，未暴露具体触发异常的原因（如静态代码块中的ExceptionInInitializerError）。
   • 修复方式：
     • 在Strimzi的KafkaConnect配置中添加日志级别调整：log4j.logger.org.apache.ignite=DEBUG；
     • 重启连接器后查看完整日志，定位静态类初始化时的具体报错，针对性解决。

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