如何捕获Flink readFile读取文件时的内部异常并跳过无效GZIP文件

当前遇到的异常本质是：异常发生在Flink FileInputFormat的输入流初始化阶段，处于ContinuousFileReaderOperator的内部调用链路中，完全在业务算子的作用域之外，因此在外层业务代码加try-catch完全无法拦截。异常触发后会直接被TaskManager判定为任务致命错误，进而触发作业终止。核心解决方向是在Flink的文件读取链路中增加坏文件的容错分支，不将无效文件的错误向上抛出。

方案一：自定义带容错逻辑的DelimitedInputFormat（改动最小，适配原有readFile逻辑） #

原有env.readFile()的底层实现就是基于DelimitedInputFormat读取文本/压缩文件，异常触发点是父类FileInputFormat.open()方法中自动加载GZIP流的逻辑。我们只需要继承原有的DelimitedInputFormat，在open阶段捕获指定的GZIP格式异常，标记当前文件无有效数据，即可让框架自动跳过该文件。

• 核心实现逻辑：

  public class LenientGzipDelimitedInputFormat extends DelimitedInputFormat<String> {
      @Override
      public void open(FileInputSplit split) throws IOException {
          try {
              // 复用父类原有的流打开、GZIP自动包装逻辑
              super.open(split);
          } catch (IOException e) {
              // 精准匹配GZIP格式校验异常，避免吞掉真实的IO故障
              if (e.getCause() instanceof ZipException 
                  && "Not in GZIP format".equals(e.getCause().getMessage())) {
                  // 重置split状态，告知框架当前split没有可读取的有效数据
                  this.pos = 0;
                  this.end = 0;
                  this.stream = null;
                  // 记录告警日志，明确跳过的无效文件路径
                  LOG.warn("跳过无效GZIP文件: {}", split.getPath());
                  return;
              }
              // 非GZIP格式类的异常正常向上抛出，避免掩盖HDFS权限、节点失联等真实故障
              throw e;
          }
      }
  
      @Override
      protected String convertRecord(String[] fields) {
          // 复用原有逻辑，将解析后的字段拼接为每行文本
          return String.join(",", fields);
      }
  }
  

• 替换原有readFile的调用方式，保留连续监控新增文件的能力：

  LenientGzipDelimitedInputFormat lenientFormat = new LenientGzipDelimitedInputFormat();
  lenientFormat.setFilePath(yourHdfsPath);
  lenientFormat.setDelimiter("\n");
  // 设置监控间隔，和原readFile的连续读配置保持一致
  lenientFormat.monitorContinuously(Duration.ofSeconds(10));
  
  DataStreamSource<String> hdfsStream = env.createInput(lenientFormat)
      .uid("lenient-hdfs-gzip-source");
  

• 注意事项：
  • 不要无差别捕获所有IOException，否则会将HDFS写入未完成、权限不足、集群节点异常等真实故障误判为无效文件跳过，造成数据丢失
  • 该自定义Format完全保留了原有readFile的自动识别GZIP、行分隔符解析、连续监控新增文件的能力，不会改变正常文件的读取逻辑
  • 针对读流中途的GZIP损坏异常，可以在nextRecord()方法中增加对应的异常捕获逻辑，处理文件传输中途损坏的场景

方案二：前置文件校验（可靠性最高，侵入性稍强） #

如果对文件校验的可靠性要求极高，可以自定义连续文件监控逻辑，在Flink将文件split发送给ReaderOperator之前，提前完成GZIP格式校验：

• 扫描到HDFS新增文件后，先打开文件输入流读取前2字节的GZIP固定魔数（0x1f8b），如果魔数不匹配，直接将该文件从待读取列表中剔除，不触发后续的流读取逻辑
• 这个方案的优点是坏文件根本不会进入实际读取链路，不会因为读坏文件占用网络IO、触发缓存异常，适合生产环境对稳定性要求极高的场景

不推荐的兜底思路 #

• 不要仅依赖Flink的无限重启策略解决这个问题：坏文件会被周期性扫描到，每次扫描到都会触发GZIP格式异常，导致作业进入无限重启的死循环，无法正常处理有效数据
• 不需要重写Flink全局的GZIP压缩流工厂：全局替换压缩工厂会影响所有作业的压缩流处理逻辑，类加载和配置成本远高于自定义InputFormat的方案

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