Flink作业写入S3 Checkpoint失败问题排查与配置咨询

1. 插件冲突：同时安装了flink-s3-fs-hadoop和flink-s3-fs-presto两个S3文件系统插件，二者会竞争处理S3相关协议，导致Flink无法正确识别s3a协议对应的实现类。
2. 配置不一致：代码和flink_properties中配置了不同的AWS凭证提供者，代码使用Hadoop S3A的TemporaryAWSCredentialsProvider，而配置文件使用Flink原生的DynamicTemporaryAWSCredentialsProvider，混用会导致凭证加载逻辑混乱。
3. 状态后端与检查点配置冲突：代码中显式设置了RocksDBStateBackend，但配置文件中又指定execution.checkpointing.storage: 'filesystem'，二者存在配置覆盖冲突，影响检查点存储路径的识别。

1. 清理冲突插件 #

仅保留与s3a协议匹配的flink-s3-fs-hadoop插件，删除flink-s3-fs-presto插件，修改Dockerfile如下：

RUN mkdir -p /opt/flink/plugins/flink-s3-fs-hadoop
ADD https://repo.maven.apache.org/maven2/org/apache/flink/flink-s3-fs-hadoop/1.18.1/flink-s3-fs-hadoop-1.18.1.jar /opt/flink/plugins/flink-s3-fs-hadoop/

2. 统一凭证与检查点配置 #

选择一种配置方式（推荐使用flink_properties集中管理），确保配置一致：

方式一：使用flink_properties配置（推荐） #

删除代码中重复的检查点与状态后端配置，仅保留必要的环境初始化，修改flink_properties：

execution.checkpointing.interval: '3 min'
execution.checkpointing.mode: 'EXACTLY_ONCE'
execution.checkpointing.timeout: '10 min'
execution.checkpointing.min-pause: '1 min'
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.externalized-checkpoint-retention: 'RETAIN_ON_CANCELLATION'

# 状态后端配置
state.backend: rocksdb
state.backend.incremental: true
state.checkpoints.dir: s3a://checkpoints

# S3A凭证配置
fs.s3a.aws.credentials.provider: 'org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider'
# 若通过环境变量传递凭证，无需手动配置access/secret/session token，Hadoop会自动读取环境变量

方式二：使用代码配置 #

删除flink_properties中重复的检查点、状态后端及S3配置，统一在代码中设置：

config = Configuration()

# S3A凭证配置
config.set_string("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider")
# 环境变量会被自动读取，也可显式设置
# config.set_string("fs.s3a.access.key", os.environ.get("AWS_ACCESS_KEY_ID"))
# config.set_string("fs.s3a.secret.key", os.environ.get("AWS_SECRET_ACCESS_KEY"))
# config.set_string("fs.s3a.session.token", os.environ.get("AWS_SESSION_TOKEN"))

stream_exec_env.get_config().set_global_job_parameters(config.to_dict())

# 检查点配置
stream_exec_env.enable_checkpointing(3 * 60 * 1000)
checkpoint_config = stream_exec_env.get_checkpoint_config()
checkpoint_config.set_checkpoint_timeout(10 * 60 * 1000)
checkpoint_config.set_max_concurrent_checkpoints(1)
checkpoint_config.set_min_pause_between_checkpoints(1 * 60 * 1000)
checkpoint_config.enable_externalized_checkpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

# RocksDB状态后端配置
checkpoint_location = "s3a://checkpoints"
rocks_db_backend = RocksDBStateBackend(checkpoint_location, True)
stream_exec_env.set_state_backend(rocks_db_backend)

3. 验证插件加载 #

确保Flink启动时能正确加载flink-s3-fs-hadoop插件，启动日志中会显示类似如下内容：

Loading plugin from /opt/flink/plugins/flink-s3-fs-hadoop/flink-s3-fs-hadoop-1.18.1.jar

• 确保Flink集群所有节点的插件目录结构一致，避免部分节点缺失插件导致的异常。
• 若使用临时凭证，需确保凭证有效期覆盖作业运行时长，避免因凭证过期导致检查点写入失败。
• 若遇到Hadoop版本兼容问题，可检查flink-s3-fs-hadoop插件依赖的Hadoop版本是否与环境匹配，必要时更换对应版本的插件。

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