升级至Flink 1.16后应用启动失败问题求助

环境信息 #

• Flink版本：1.16.0
• 作业启动命令：

./bin/flink run-application --target kubernetes-application -Dkubernetes.cluster-id=sqs-signal-ingress-cluster -Dkubernetes.namespace=dev-sqs -Dkubernetes.jobmanager.service-account=flink-service-account -Dkubernetes.container.image=acrccsdev.azurecr.io/ubs-changes-oct21-sqs-signal-ingress:2022-10-21.11_27294_sqs-signal-ingress_ubs-changes-oct21 -Dkubernetes.container.image.pull-secrets=dev-ccs-acr local:///opt/flink/usrlib/signal-ingress-job.jar

报错详情 #

Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.apache.flink.connector.kafka.sink.KafkaSink]: Factory method 'kafkaSinkFsmStates' threw exception; nested exception is java.lang.LinkageError: loader constraint violation: loader org.apache.flink.util.ChildFirstClassLoader @10feca44 wants to load class org.apache.kafka.clients.producer.ProducerRecord. A different class with the same name was previously loaded by 'app'. (org.apache.kafka.clients.producer.ProducerRecord is in unnamed module of loader 'app')
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[signal-ingress-job.jar:0.0.1-SNAPSHOT]
    at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:653) ~[signal-ingress-job.jar:0.0.1-SNAPSHOT]
    ... 53 common frames omitted
Caused by: java.lang.LinkageError: loader constraint violation: loader org.apache.flink.util.ChildFirstClassLoader @10feca44 wants to load class org.apache.kafka.clients.producer.ProducerRecord. A different class with the same name was previously loaded by 'app'. (org.apache.kafka.clients.producer.ProducerRecord is in unnamed module of loader 'app')
    at java.base/java.lang.ClassLoader.defineClass1(Native Method) ~[na:na]
    at java.base/java.lang.ClassLoader.defineClass(Unknown Source) ~[na:na]

解决方案 #

这个错误是因为Kafka客户端类被重复加载：应用类加载器先加载了ProducerRecord，之后Flink的ChildFirstClassLoader又尝试加载同一类但版本/来源不同，导致类加载冲突。

具体解决步骤： #

• 排查并修复依赖冲突：
  检查作业的构建文件（如pom.xml），将Kafka客户端依赖的scope设为provided，因为Flink的Kafka连接器已内置兼容版本，无需打包进作业jar：

  <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>${kafka.version}</version>
      <scope>provided</scope>
  </dependency>
  

  同时确保Flink Kafka连接器依赖与Flink版本严格匹配：

  <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka</artifactId>
      <version>${flink.version}</version>
  </dependency>
  

• 调整Flink类加载策略：
  在启动命令中追加类加载配置，强制Kafka相关类由父类加载器优先加载，避免冲突：

  -Dclassloader.resolve-order=parent-first -Dflink.classloader.parent-first-patterns.additional=org.apache.kafka
  

  修改后的完整启动命令示例：

  ./bin/flink run-application --target kubernetes-application -Dkubernetes.cluster-id=sqs-signal-ingress-cluster -Dkubernetes.namespace=dev-sqs -Dkubernetes.jobmanager.service-account=flink-service-account -Dkubernetes.container.image=acrccsdev.azurecr.io/ubs-changes-oct21-sqs-signal-ingress:2022-10-21.11_27294_sqs-signal-ingress_ubs-changes-oct21 -Dkubernetes.container.image.pull-secrets=dev-ccs-acr -Dclassloader.resolve-order=parent-first -Dflink.classloader.parent-first-patterns.additional=org.apache.kafka local:///opt/flink/usrlib/signal-ingress-job.jar
  

• 清理镜像冗余依赖：
  用mvn dependency:tree检查依赖树，或用jar tf signal-ingress-job.jar查看jar包内容，确认Kafka客户端类未被打包进作业jar。如果存在冗余依赖，重新构建jar排除相关依赖。

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