在Flink中创建Kafka Consumer时遇java.lang.NoClassDefFoundError(CheckpointedRestoring)

这个错误本质上是类路径中缺失了包含CheckpointedRestoring类的Flink依赖，或是Flink核心依赖与Kafka连接器依赖版本不匹配导致的。下面是具体的排查和解决步骤：

1. 严格对齐Flink核心与Kafka连接器的版本 #

Flink的Kafka连接器和核心框架版本必须完全一致，版本不兼容是这类类找不到问题的头号原因。

比如你用的是Flink 1.17.x，那Kafka连接器依赖也得是flink-connector-kafka-1.17.x版本。给你个Maven依赖的正确示例：

<!-- Flink核心依赖 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>1.17.1</version>
    <scope>provided</scope>
</dependency>

<!-- Flink Kafka连接器依赖 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.17.1</version>
</dependency>

2. 确认核心依赖完整引入 #

CheckpointedRestoring属于Flink streaming核心模块，要确保flink-streaming-java（Scala项目用flink-streaming-scala）已经正确加入项目依赖，且运行时能被加载到：

• IDE运行时：检查项目依赖库中是否存在这个类；
• 打包运行时：用Maven Shade插件或Flink官方推荐的Fat Jar打包方式，避免依赖丢失。

3. 排查集群运行时的类路径问题 #

如果是在Flink集群上运行：

• 检查集群lib目录下的Flink核心Jar版本，是否和你项目打包时用的版本一致；
• 用flink run提交作业时，确保你的Jar是包含所有必要依赖的Fat Jar，或者集群环境已经预装了对应依赖。

4. 避免使用过时API #

旧版的FlinkKafkaConsumerAPI已经被废弃，而CheckpointedRestoring可能是在新版本中新增的类。建议使用Flink 1.13+推荐的KafkaSource新API，示例代码如下：

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KafkaConsumerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("your-topic")
                .setGroupId("flink-consumer-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new org.apache.flink.api.common.serialization.SimpleStringSchema())
                .build();

        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
                .print();

        env.execute("Flink Kafka Consumer Job");
    }
}

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