Flink 1.11升级至1.15.1后，配置FileSink在S3生成.gz文件遇UnsupportedOperationException问题求助

这个错误的核心原因很明确：你当前使用的Hadoop文件系统实现里，HadoopRecoverableWriter仅适配HDFS，S3并不兼容这种可恢复写入机制。下面是具体的解决步骤和代码调整方案：

一、核心问题拆解 #

FileSink默认依赖**可恢复写入器（RecoverableWriter）**来实现精确一次语义，但Hadoop的S3客户端（或你当前配置的S3文件系统实现）并没有适配这个写入器。Flink 1.15+对S3的支持需要专门的文件系统配置，不能直接复用HDFS的逻辑。

二、解决方案步骤 #

1. 配置正确的S3文件系统实现 #

你需要选择以下两种方式之一来适配Flink的S3文件系统：

• 方式一：使用Flink原生S3文件系统
  这种方式无需依赖Hadoop的S3A客户端，直接用Flink官方维护的S3插件。在flink-conf.yaml中添加配置：

  fs.s3.impl: org.apache.flink.fs.s3.common.hadoop.HadoopS3FileSystem
  fs.s3.access-key: 你的S3访问密钥
  fs.s3.secret-key: 你的S3密钥
  # 可选：如果是S3兼容存储，配置endpoint
  # fs.s3.endpoint: https://你的存储节点地址
  

  同时确保项目依赖包含Flink S3相关组件：

  <!-- Maven依赖示例 -->
  <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-filesystem</artifactId>
      <version>1.15.1</version>
  </dependency>
  <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-s3-fs-hadoop</artifactId>
      <version>1.15.1</version>
  </dependency>
  

• 方式二：使用Hadoop S3A并启用可恢复写入
  若坚持用Hadoop的S3A客户端，需要在flink-conf.yaml中开启S3A的可恢复写入支持：

  fs.s3a.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
  fs.s3a.access.key: 你的S3访问密钥
  fs.s3a.secret.key: 你的S3密钥
  # 启用S3A分段写入与可恢复特性
  fs.s3a.multipart.enabled: true
  fs.s3a.multipart.size: 134217728 # 128MB，可按需调整
  fs.s3a.recoverable.write.enable: true
  

  注意：这种方式要求Hadoop版本在3.2及以上，确保与Flink 1.15.1兼容。

2. 调整FileSink代码与作业配置 #

你的代码中使用了OnCheckpointRollingPolicy，必须确保作业已启用Checkpoint机制：

// 在StreamExecutionEnvironment中开启Checkpoint
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000); // 每60秒触发一次Checkpoint
env.getCheckpointConfig().setCheckpointStorage("s3://你的Checkpoint存储路径");

另外，你可以简化Gzip编码逻辑，Flink提供了现成的GzipWriterFactory，无需手动实现Encoder：

return FileSink.forRowFormat(new Path(outputBasePath),
        // 用Jackson序列化并自动适配Gzip
        new SimpleStringEncoder<T>(element -> OBJECT_MAPPER.writeValueAsString(element), StandardCharsets.UTF_8))
        .withBucketAssigner(new BasePathBucketAssigner<>())
        .withRollingPolicy(OnCheckpointRollingPolicy.build())
        // 指定输出文件的.gz后缀
        .withOutputFileConfig(OutputFileConfig.builder()
                .withPartSuffix(".gz")
                .build())
        .withWriterFactory(new GzipWriterFactory<>())
        .build();

如果仍需手动实现Encoder，建议用try-with-resources确保流正确关闭：

return FileSink.forRowFormat(new Path(outputBasePath), new Encoder<T>() {
    @Override
    public void encode(T record, OutputStream stream) throws IOException {
        GzipParameters params = new GzipParameters();
        params.setCompressionLevel(Deflater.BEST_COMPRESSION);
        // 自动关闭Gzip流
        try (GzipCompressorOutputStream out = new GzipCompressorOutputStream(stream, params)) {
            OBJECT_MAPPER.writeValue(out, record);
        }
    }
})
.withBucketAssigner(new BasePathBucketAssigner<>())
.withRollingPolicy(OnCheckpointRollingPolicy.build())
.withOutputFileConfig(OutputFileConfig.builder().withPartSuffix(".gz").build())
.build();

3. 验证权限与路径配置 #

确保Flink作业拥有以下权限：

• 写入outputBasePath的权限
• 若使用Checkpoint，写入Checkpoint存储路径的权限
• 确认S3路径格式正确（如s3://bucket/path或s3a://bucket/path，与配置的文件系统实现对应）

三、常见排查点 #

• 检查依赖版本是否与Flink 1.15.1兼容，避免版本冲突
• 确认flink-conf.yaml中的文件系统配置无拼写错误
• 若使用S3兼容存储，需确保endpoint配置正确

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