Spark Streaming foreachRDD任务异常仍提交Kafka偏移量问题咨询

首先明确你的实现核心问题出在Spark的Task异常传递机制和你当前代码缺少执行结果校验两个层面，具体如下：

核心问题点 #

• Task重试机制掩盖了单次异常，未触发Driver端作业失败
  Spark 2.1.1版本默认配置spark.task.maxFailures=4，单个Task允许最多失败重试4次，只要重试次数未耗尽、最终Task执行成功，Driver端的doTheThing(rdd)调用就不会抛出异常，会直接执行后续的偏移量提交逻辑。如果你的NeedToAbortSomeException属于业务层面不可重试的异常，单次抛出就应该终止整批处理，默认的重试机制会直接违背你的预期。
• Executor端异常未正确传递终止信号到Driver
  你在foreachPartition内抛出的异常是Executor侧的Task级异常，只有当整批作业最终被标记为失败（重试超过最大次数）时，才会向Driver侧的foreachRDD代码块抛出异常。如果没有额外的异常捕获逻辑，一旦作业最终执行成功（哪怕中间有单次Task失败），偏移量都会被正常提交。
• 缺少执行结果校验逻辑
  你当前的代码默认doTheThing执行完成就等于处理成功，没有主动校验所有分区是否全部正常处理完成，极端情况下哪怕部分Task执行失败被框架静默处理，也会触发偏移量提交。

修复方案 #

你可以根据业务需求选择以下一种或多种组合修复：

1. 调整Task失败重试策略
   如果业务要求出现NeedToAbortSomeException就必须终止整批处理、不允许重试，可以将自定义异常标记为Spark不可重试异常，或者直接配置spark.task.maxFailures=1，确保单次Task失败就直接标记整批作业失败。
2. 在Driver端增加异常捕获逻辑，异常时跳过偏移量提交
   修改后的代码示例：

kafkaStream.foreachRDD { rdd =>
        if (!rdd.isEmpty()) {
          val offsets = offsetStore.getOffsetsFromRDD(rdd)        
          var processSuccess = false
          try {
              doTheThing(rdd)
              processSuccess = true
          } catch {
              case e: Exception => 
                  // 可按需增加告警、作业终止逻辑
                  log.error("当前批次处理失败，不提交偏移量", e)
          }
          if (processSuccess) {
              offsetStore.persistOffsets(offsets)          
          }
        }
 }

3. 增加分区处理结果校验
   可以在doTheThing中累加各分区处理成功的标记，最终和RDD的总分区数对比，确认所有分区都处理成功后再提交偏移量，避免框架静默吞掉异常的极端情况。

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