Cuando se utiliza Kafka con Spark Streaming y se almacenan los offsets en Zookeeper, es común enfrentarse al error kafka.common.OffsetOutOfRangeException tras un periodo de inactividad. Este problema ocurre cuando Kafka ha ejecutado su política de retención y limpieza de logs, eliminando los mensajes más antiguos. Al intentar reanudar el consumo desde un offset guardado que ya no existe en el broker, la petición falla.

Causa del problema

Kafka periódicamente purga los segmentos de log más viejos según la configuración de retención. Si un consumidor guardó su posición (offset) en Zookeeper antes de la purga y luego intenta reconectar, el offset solicitado será menor que el offset inicial (earliest offset) disponible actualmente en la partición. Como resulatdo, Kafka lanza una excepción de fuera de rango.

Estrategia de solución

La solución consiste en implementar un mecanismo de validación de offsets antes de comenzar el flujo de Spark Streaming. Para cada partición, se debe comprboar si el offset almacenado en Zookeeper se encuentra dentro del rango válido actual [earliest_offset, latest_offset]. Si el offset guardado es menor que el inicial o mayor que el final, se debe corregir y ajustar al offset inicial disponible.

Implementación del coordinador de offsets

La siguiente clase KafkaOffsetCoordinator encapsula la lógica para leer, validar, corregir y guardar offsets utilizando Zookeeper y la API de consumidores de Kafka.

import kafka.utils.{ZKGroupTopicDirs, ZkUtils}
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer, NoOffsetForPartitionException}
import org.apache.kafka.common.TopicPartition
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies}
import org.slf4j.LoggerFactory

import scala.collection.mutable
import scala.reflect.ClassTag
import scala.util.Try

class KafkaOffsetCoordinator(zkQuorum: String, kafkaConfig: Map[String, Object]) extends Serializable {
  
  @transient private lazy val logger = LoggerFactory.getLogger(getClass)
  
  private val (zkClient, zkConn) = ZkUtils.createZkClientAndConnection(zkQuorum, 10000, 10000)
  private val zkHelper = new ZkUtils(zkClient, zkConn, false)

  private def buildConsumer(resetPolicy: String): KafkaConsumer[String, Array[Byte]] = {
    val props = mutable.Map[String, Object]() ++= kafkaConfig
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, resetPolicy)
    new KafkaConsumer[String, Array[Byte]](props)
  }

  def fetchStartingOffsets(topics: Iterable[String]): Map[TopicPartition, Long] = {
    val kafkaConsumer = buildConsumer("earliest")
    kafkaConsumer.subscribe(topics)
    
    try { kafkaConsumer.poll(0) } catch {
      case _: NoOffsetForPartitionException => 
    }
    
    val assignedPartitions = kafkaConsumer.assignment().toSet
    kafkaConsumer.pause(assignedPartitions)
    kafkaConsumer.seekToBeginning(assignedPartitions)
    
    val startingPoints = assignedPartitions.map(tp => tp -> kafkaConsumer.position(tp)).toMap
    kafkaConsumer.close()
    startingPoints
  }

  def fetchEndingOffsets(topics: Iterable[String]): Map[TopicPartition, Long] = {
    val kafkaConsumer = buildConsumer("latest")
    kafkaConsumer.subscribe(topics)
    
    try { kafkaConsumer.poll(0) } catch {
      case _: NoOffsetForPartitionException => 
    }
    
    val assignedPartitions = kafkaConsumer.assignment().toSet
    kafkaConsumer.pause(assignedPartitions)
    kafkaConsumer.seekToEnd(assignedPartitions)
    
    val endingPoints = assignedPartitions.map(tp => tp -> kafkaConsumer.position(tp)).toMap
    kafkaConsumer.close()
    endingPoints
  }

  def loadOffsets(topicList: Seq[String], consumerGroup: String): Map[TopicPartition, Long] = {
    val offsetMap = mutable.HashMap.empty[TopicPartition, Long]
    val topicPartitionInfo = zkHelper.getPartitionsForTopics(topicList)
    
    topicPartitionInfo.foreach { case (topic, partitions) =>
      val zkDirs = new ZKGroupTopicDirs(consumerGroup, topic)
      partitions.foreach { partition =>
        val path = s"${zkDirs.consumerOffsetDir}/$partition"
        val offsetAttempt = Try {
          val (data, _) = zkHelper.readData(path)
          if (data != null) {
            logger.info(s"Topic: $topic, Partition: $partition, Offset: $data, Path: $path")
            offsetMap.put(new TopicPartition(topic, partition.toInt), data.toLong)
          }
        }
        if (offsetAttempt.isFailure) {
          logger.warn(s"No previous offset found in ZK for $topic-$partition. Fetching from Kafka.")
          offsetMap.put(new TopicPartition(topic, partition.toInt), 0L)
        }
      }
    }
    
    val lowerBounds = fetchStartingOffsets(topicList)
    val upperBounds = fetchEndingOffsets(topicList)
    
    offsetMap.foreach { case (tp, storedOffset) =>
      val minOffset = lowerBounds.getOrElse(tp, 0L)
      val maxOffset = upperBounds.getOrElse(tp, Long.MaxValue)
      
      if (storedOffset > maxOffset || storedOffset < minOffset) {
        logger.warn(s"Offset $storedOffset for $tp is out of range [$minOffset, $maxOffset]. Resetting to $minOffset.")
        offsetMap(tp) = minOffset
      }
    }
    
    offsetMap.toMap
  }

  def createStream[K: ClassTag, V: ClassTag](ssc: StreamingContext, topicList: Seq[String]): InputDStream[ConsumerRecord[K, V]] = {
    val groupId = kafkaConfig("group.id").toString
    val validatedOffsets = loadOffsets(topicList, groupId)
    KafkaUtils.createDirectStream[K, V](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[K, V](topicList, kafkaConfig, validatedOffsets)
    )
  }

  def commitOffsets[K, V](rdd: RDD[ConsumerRecord[K, V]], useEndOffset: Boolean = true): Unit = {
    val groupId = kafkaConfig("group.id").toString
    val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    ranges.foreach { range =>
      val zkDirs = new ZKGroupTopicDirs(groupId, range.topic)
      val path = s"${zkDirs.consumerOffsetDir}/${range.partition}"
      val commitValue = if (useEndOffset) range.untilOffset else range.fromOffset
      zkHelper.updatePersistentPath(path, commitValue.toString)
      logger.debug(s"Committed offset $commitValue for ${range.topic}-${range.partition} at $path")
    }
  }
}

Uso de la clase en Spark Streaming

Para integrar este coordinador en tu aplicación, simplemente instánciala con los parámetros de Zookeeper y Kafka, utilízala para crear el flujo directo de datos y confirma los offsets procesados al final de cada lote RDD.

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "localhost:9092",
  "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "value.deserializer" -> "org.apache.kafka.common.serialization.ByteArrayDeserializer",
  "group.id" -> "mi_grupo_consumidor"
)

val coordinator = new KafkaOffsetCoordinator("localhost:2181", kafkaParams)

val stream = coordinator.createStream[String, Array[Byte]](ssc, Seq("mi_topic"))

stream.foreachRDD { rdd =>
  // Procesamiento de datos...
  
  // Guardar progreso
  coordinator.commitOffsets(rdd)
}