Spark Streaming任务Kafka分区重分配后失败：报无当前分区分配

这个问题我做Spark-Kafka对接时也碰到过，本质是分区重新分配（Rebalance）后，你还在尝试提交已经不属于当前消费者的旧分区偏移量，才会抛出No current assignment for partition topic1这个错误。

问题原因分析 #

从你提供的日志能看到，Consumer Group已经触发了Rebalance：撤销了旧分区[TOPIC_NAME-3, TOPIC_NAME-5, TOPIC_NAME-4]，然后重新加入组分配新分区。但你的代码在foreachRDD里直接获取当前RDD的offsetRanges就提交，完全没考虑Rebalance的情况——当Rebalance发生时，这个RDD对应的旧分区已经被回收，当前消费者实例已经没有这些分区的权限了，这时候调用commitAsync自然会报错。

解决方案 #

方案1：捕获特定异常，跳过无效分区的提交 #

修改偏移量提交逻辑，针对Rebalance导致的分区未分配异常做特殊处理，跳过这些无效的提交请求。同时要确保数据处理完成后再提交偏移量，避免丢数：

try {
  messages.foreachRDD { rdd =>
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    
    // 第一步：先处理RDD中的业务数据，确保数据处理完成再提交偏移量
    rdd.foreachPartition { partitionIter =>
      // 这里写你的业务处理逻辑，比如解析Kafka消息、计算、写入存储等
      partitionIter.foreach { msg =>
        // process message
      }
    }
    
    // 第二步：尝试提交偏移量，处理Rebalance异常
    try {
      messages.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges, new OffsetCommitCallback {
        override def onComplete(offsets: util.Map[TopicPartition, OffsetAndMetadata], exception: Exception): Unit = {
          if (exception != null) {
            exception match {
              // 匹配Rebalance导致的分区未分配异常
              case e: IllegalStateException if e.getMessage.contains("No current assignment for partition") =>
                println(s"Rebalance触发，跳过已撤销分区的偏移量提交: ${e.getMessage}")
              // 其他异常正常打印堆栈
              case _ =>
                e.printStackTrace()
            }
          }
        }
      })
    } catch {
      case e: IllegalStateException if e.getMessage.contains("No current assignment for partition") =>
        println(s"检测到Rebalance，旧分区已失效，跳过提交: ${e.getMessage}")
      case e: Throwable =>
        e.printStackTrace()
    }
  }
} catch {
  case e: Throwable => e.printStackTrace()
}

方案2：改用Spark Checkpoint自动管理偏移量 #

如果你的业务场景不需要强绑定“处理完成再提交”的逻辑，可以直接让Spark通过Checkpoint来自动保存和恢复偏移量，这样就不用手动调用commitAsync，从根源上避免这个问题：

// 初始化StreamingContext时设置Checkpoint目录
val ssc = new StreamingContext(sparkConf, Seconds(5))
ssc.checkpoint("/path/to/checkpoint/dir")

// 创建Direct Stream时，无需手动提交偏移量，Spark会自动管理
val messages = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams)
)

// 正常处理业务逻辑即可
messages.foreachRDD { rdd =>
  // 业务处理代码
}

额外优化建议 #

• 检查你的Kafka参数：session.timeout.ms和request.timeout.ms不要设置得太小，否则消费者容易因超时被踢出组，引发频繁Rebalance，增加报错概率。
• 如果必须手动提交偏移量，尽量保证每个批次的处理时间小于批次间隔，避免消费者长时间未心跳导致Rebalance。

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