spark streaming 接收kafka消息之四 -- 運行在 worker 上的 receiver

使用分布式receiver來獲取數據
使用 WAL 來實現 At least once 操作：
conf.set("spark.streaming.receiver.writeAheadLog.enable","true") // 開啟 WAL
// 1、At most once - 每條數據最多被處理一次（0次或1次），這種語義下會出現數據丟失的問題；
// 2、At least once - 每條數據最少被處理一次 (1次或更多)，這個不會出現數據丟失，但是會出現數據重復；
// 3、Exactly once - 每條數據只會被處理一次，沒有數據會丟失，並且沒有數據會被多次處理，這種語義是大家最想要的，但是也是最難實現的。

如果不做容錯，將會帶來數據丟失，因為Receiver一直在接收數據，在其沒有處理的時候（已通知zk數據接收到），Executor突然掛掉(或是driver掛掉通知executor關閉)，緩存在內存中的數據就會丟失。因為這個問題，Spark1.2開始加入了WAL（Write ahead log）開啟 WAL，將receiver獲取數據的存儲級別修改為StorageLevel. MEMORY_AND_DISK_SER_2

// 缺點，不能自己維護消費 topic partition 的 offset
// 優點，開啟 WAL，來確保 exactly-once 語義
val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](
    ssc,kafkaParams,map,StorageLevel.MEMORY_AND_DISK_SER_2)

從Kafka 中讀取數據

Driver 規划 receiver 運行的信息

org.apache.spark.streaming.StreamingContext#start中啟動了 JobScheduler實例

// private[streaming] val scheduler = new JobScheduler(this)

// Start the streaming scheduler in a new thread, so that thread local properties
// like call sites and job groups can be reset without affecting those of the
// current thread.
ThreadUtils.runInNewThread("streaming-start") { // 單獨的一個daemon線程運行函數題
  sparkContext.setCallSite(startSite.get)
  sparkContext.clearJobGroup()
  sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false")
// 執行start 方法
  scheduler.start()
}
state = StreamingContextState.ACTIVE

 

org.apache.spark.streaming.scheduler.JobScheduler#start 源碼如下：

def start(): Unit = synchronized {
  if (eventLoop != null) return // scheduler has already been started

  logDebug("Starting JobScheduler")
  eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
    override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)

    override protected def onError(e: Throwable): Unit = reportError("Error in job scheduler", e)
  }
  eventLoop.start()

  // attach rate controllers of input streams to receive batch completion updates
  for {
    inputDStream <- ssc.graph.getInputStreams
    rateController <- inputDStream.rateController
  } ssc.addStreamingListener(rateController)

  listenerBus.start(ssc.sparkContext)
  receiverTracker = new ReceiverTracker(ssc)
  inputInfoTracker = new InputInfoTracker(ssc)
  receiverTracker.start()
  jobGenerator.start()
  logInfo("Started JobScheduler")
}

 

 

ReceiverTracker 的類聲明如下：

This class manages the execution of the receivers of ReceiverInputDStreams. Instance of this class must be created after all input streams have been added and StreamingContext.start() has been called because it needs the final set of input streams at the time of instantiation.
此類負責執行ReceiverInputDStreams的receiver。必須在添加所有輸入流並調用StreamingContext.start（）之后創建此類的實例，因為它在實例化時需要最終的輸入流集。

 

其 start 方法如下：

/** Start the endpoint and receiver execution thread. */
def start(): Unit = synchronized {
  if (isTrackerStarted) {
    throw new SparkException("ReceiverTracker already started")
  }

  if (!receiverInputStreams.isEmpty) {
// 建立rpc endpoint
    endpoint = ssc.env.rpcEnv.setupEndpoint( // 注意，這是一個driver的 endpoint
      "ReceiverTracker", new ReceiverTrackerEndpoint(ssc.env.rpcEnv))
// driver節點上發送啟動 receiver 命令
    if (!skipReceiverLaunch) launchReceivers()
    logInfo("ReceiverTracker started")
    trackerState = Started
  }
}

/**
 * Get the receivers from the ReceiverInputDStreams, distributes them to the
 * worker nodes as a parallel collection, and runs them.
 */
// 從ReceiverInputDStreams 獲取到 receivers，然后將它們分配到不同的 worker 節點並運行它們。
private def launchReceivers(): Unit = {
  val receivers = receiverInputStreams.map(nis => {
// 未啟用WAL 是KafkaReceiver，啟動WAL后是ReliableKafkaReceiver
    val rcvr = nis.getReceiver()
    rcvr.setReceiverId(nis.id)
    rcvr
  })
  // 運行一個簡單的應用來確保所有的salve node都已經啟動起來，避免所有的 receiver 任務都在同一個local node上
  runDummySparkJob()

  logInfo("Starting " + receivers.length + " receivers")
  endpoint.send(StartAllReceivers(receivers)) // 發送請求driver 轉發 啟動 receiver 的命令
}

Driver 端StartAllReceivers 的處理代碼如下：

override def receive: PartialFunction[Any, Unit] = {
  // Local messages
  case StartAllReceivers(receivers) =>
// schduleReceiver
    val scheduledLocations = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
    for (receiver <- receivers) {
      val executors = scheduledLocations(receiver.streamId)
      updateReceiverScheduledExecutors(receiver.streamId, executors)
      receiverPreferredLocations(receiver.streamId) = receiver.preferredLocation
      startReceiver(receiver, executors)
    }
……
}

 

getExecutors源碼如下：

/**
 * Get the list of executors excluding driver
 */
// 如果是 local 模式，返回 本地線程； 如果是 yarn 模式，返回 非driver 節點上的 excutors
private def getExecutors: Seq[ExecutorCacheTaskLocation] = {
  if (ssc.sc.isLocal) { // 如果在 local 模式下運行
    val blockManagerId = ssc.sparkContext.env.blockManager.blockManagerId
    Seq(ExecutorCacheTaskLocation(blockManagerId.host, blockManagerId.executorId))
  } else { // 在 yarn 模式下，過濾掉 driver 的 executor
    ssc.sparkContext.env.blockManager.master.getMemoryStatus.filter { case (blockManagerId, _) =>
      blockManagerId.executorId != SparkContext.DRIVER_IDENTIFIER // Ignore the driver location
    }.map { case (blockManagerId, _) =>
      ExecutorCacheTaskLocation(blockManagerId.host, blockManagerId.executorId)
    }.toSeq
  }
}

org.apache.spark.streaming.scheduler.ReceiverSchedulingPolicy#scheduleReceivers的解釋如下：

Try our best to schedule receivers with evenly distributed. However, if the preferredLocations of receivers are not even, we may not be able to schedule them evenly because we have to respect them. Here is the approach to schedule executors:
First, schedule all the receivers with preferred locations (hosts), evenly among the executors running on those host.
Then, schedule all other receivers evenly among all the executors such that overall distribution over all the receivers is even.
This method is called when we start to launch receivers at the first time.
該方法就是確保receiver 能夠在worker node 上均勻分布的。遵循以下兩個原則：
1.使用 preferred location 分配 receiver 到這些node 上
2.將其他的未分配的receiver均勻分布均勻分布到 每一個 worker node 上 

org.apache.spark.streaming.scheduler.ReceiverTracker#updateReceiverScheduledExecutors 負責更新receiverid 和 receiver info 的映射關系，源碼如下：

private def updateReceiverScheduledExecutors(
    receiverId: Int, scheduledLocations: Seq[TaskLocation]): Unit = {
  val newReceiverTrackingInfo = receiverTrackingInfos.get(receiverId) match {
    case Some(oldInfo) =>
      oldInfo.copy(state = ReceiverState.SCHEDULED,
        scheduledLocations = Some(scheduledLocations))
    case None =>
      ReceiverTrackingInfo(
        receiverId,
        ReceiverState.SCHEDULED,
        Some(scheduledLocations),
        runningExecutor = None)
  }
  receiverTrackingInfos.put(receiverId, newReceiverTrackingInfo)
}

 

Driver 發送分布式啟動receiver job

startReceiver 負責啟動 receiver，源碼如下：

/**
 * Start a receiver along with its scheduled executors
 */
private def startReceiver(
    receiver: Receiver[_],
    scheduledLocations: Seq[TaskLocation]): Unit = {
  def shouldStartReceiver: Boolean = {
    // It's okay to start when trackerState is Initialized or Started
    !(isTrackerStopping || isTrackerStopped)
  }

  val receiverId = receiver.streamId
  if (!shouldStartReceiver) {
    onReceiverJobFinish(receiverId)
    return
  }

  val checkpointDirOption = Option(ssc.checkpointDir)
  val serializableHadoopConf =
    new SerializableConfiguration(ssc.sparkContext.hadoopConfiguration)

// 在 worker node 上啟動 receiver 的方法
  val startReceiverFunc: Iterator[Receiver[_]] => Unit =
    (iterator: Iterator[Receiver[_]]) => {
      if (!iterator.hasNext) {
        throw new SparkException(
          "Could not start receiver as object not found.")
      }
      if (TaskContext.get().attemptNumber() == 0) {
        val receiver = iterator.next()
        assert(iterator.hasNext == false)
        val supervisor = new ReceiverSupervisorImpl(
          receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
        supervisor.start()
        supervisor.awaitTermination()
      } else {
        // It's restarted by TaskScheduler, but we want to reschedule it again. So exit it.
      }
    }

  // Create the RDD using the scheduledLocations to run the receiver in a Spark job
  val receiverRDD: RDD[Receiver[_]] =
    if (scheduledLocations.isEmpty) {
      ssc.sc.makeRDD(Seq(receiver), 1)
    } else {
      val preferredLocations = scheduledLocations.map(_.toString).distinct
      ssc.sc.makeRDD(Seq(receiver -> preferredLocations))
    }
  receiverRDD.setName(s"Receiver $receiverId")
  ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
  ssc.sparkContext.setCallSite(Option(ssc.getStartSite()).getOrElse(Utils.getCallSite()))
  // 提交分布式receiver 啟動任務
  val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
    receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
  // We will keep restarting the receiver job until ReceiverTracker is stopped
  future.onComplete {
    case Success(_) =>
      if (!shouldStartReceiver) {
        onReceiverJobFinish(receiverId)
      } else {
        logInfo(s"Restarting Receiver $receiverId")
        self.send(RestartReceiver(receiver))
      }
    case Failure(e) =>
      if (!shouldStartReceiver) {
        onReceiverJobFinish(receiverId)
      } else {
        logError("Receiver has been stopped. Try to restart it.", e)
        logInfo(s"Restarting Receiver $receiverId")
        self.send(RestartReceiver(receiver))
      }
  }(submitJobThreadPool)
  logInfo(s"Receiver ${receiver.streamId} started")
}

Worker節點啟動 receiver監管服務

org.apache.spark.streaming.receiver.ReceiverSupervisorImpl#ReceiverSupervisorImpl 的 start 方法如下：

/** Start the supervisor */
def start() {
  onStart()
  startReceiver()
}
override protected def onStart() { // 啟動 BlockGenerator 服務
  registeredBlockGenerators.foreach { _.start() }
}
// startReceiver 方法如下：
/** Start receiver */
def startReceiver(): Unit = synchronized {
  try {
    if (onReceiverStart()) { // 注冊receiver 成功
      logInfo("Starting receiver")
      receiverState = Started
      receiver.onStart() // 啟動 receiver
      logInfo("Called receiver onStart")
    } else {
      // The driver refused us
      stop("Registered unsuccessfully because Driver refused to start receiver " + streamId, None)
    }
  } catch {
    case NonFatal(t) =>
      stop("Error starting receiver " + streamId, Some(t))
  }
}

 

注冊 receiver 到 driver節點

override protected def onReceiverStart(): Boolean = {
  val msg = RegisterReceiver(
    streamId, receiver.getClass.getSimpleName, host, executorId, endpoint)
  trackerEndpoint.askWithRetry[Boolean](msg)
}

 

簡單描述一下driver 端做的事情，主要負責將其納入到org.apache.spark.streaming.scheduler.ReceiverTracker 的管理中來，具體streamid 和 ReceiverTrackingInfo 的映射關系保存在receiverTrackingInfos中。

org.apache.spark.streaming.scheduler.ReceiverTracker#registerReceiver關鍵代碼如下：

val name = s"${typ}-${streamId}"
val receiverTrackingInfo = ReceiverTrackingInfo(
  streamId,
  ReceiverState.ACTIVE,
  scheduledLocations = None,
  runningExecutor = Some(ExecutorCacheTaskLocation(host, executorId)),
  name = Some(name),
  endpoint = Some(receiverEndpoint))
receiverTrackingInfos.put(streamId, receiverTrackingInfo)
listenerBus.post(StreamingListenerReceiverStarted(receiverTrackingInfo.toReceiverInfo))

 

啟動 receiver 線程

由於我們啟用了 WAL， 所以 這里的receiver 是ReliableKafkaReceiver 的實例
receiver.onStart 即 org.apache.spark.streaming.kafka.ReliableKafkaReceiver#onStart, 源碼如下：

override def onStart(): Unit = {
  logInfo(s"Starting Kafka Consumer Stream with group: $groupId")

  // Initialize the topic-partition / offset hash map.
// 1. 負責維護消費的 topic-partition 和 offset 的映射關系
  topicPartitionOffsetMap = new mutable.HashMap[TopicAndPartition, Long]

  // Initialize the stream block id / offset snapshot hash map.
// 2. 負責維護 block-id 和 partition-offset 之間的映射關系
  blockOffsetMap = new ConcurrentHashMap[StreamBlockId, Map[TopicAndPartition, Long]]()

  // Initialize the block generator for storing Kafka message.
// 3. 負責保存 kafka message 的 block generator，入參是GeneratedBlockHandler 實例，這是一個負責監聽 block generator事件的一個監聽器
// Generates batches of objects received by a org.apache.spark.streaming.receiver.Receiver and puts them into appropriately named blocks at regular intervals. This class starts two threads, one to periodically start a new batch and prepare the previous batch of as a block, the other to push the blocks into the block manager. 
  blockGenerator = supervisor.createBlockGenerator(new GeneratedBlockHandler)
  // 4. 關閉consumer 自動提交 offset 選項
// auto_offset_commit 應該是 false
  if (kafkaParams.contains(AUTO_OFFSET_COMMIT) && kafkaParams(AUTO_OFFSET_COMMIT) == "true") {
    logWarning(s"$AUTO_OFFSET_COMMIT should be set to false in ReliableKafkaReceiver, " +
      "otherwise we will manually set it to false to turn off auto offset commit in Kafka")
  }

  val props = new Properties()
  kafkaParams.foreach(param => props.put(param._1, param._2))
  // Manually set "auto.commit.enable" to "false" no matter user explicitly set it to true,
  // we have to make sure this property is set to false to turn off auto commit mechanism in Kafka.
  props.setProperty(AUTO_OFFSET_COMMIT, "false")

  val consumerConfig = new ConsumerConfig(props)

  assert(!consumerConfig.autoCommitEnable)

  logInfo(s"Connecting to Zookeeper: ${consumerConfig.zkConnect}")
// 5. 初始化 consumer 對象
// consumerConnector 是ZookeeperConsumerConnector的實例
  consumerConnector = Consumer.create(consumerConfig)
  logInfo(s"Connected to Zookeeper: ${consumerConfig.zkConnect}")
  // 6. 初始化zookeeper 的客戶端
  zkClient = new ZkClient(consumerConfig.zkConnect, consumerConfig.zkSessionTimeoutMs,
    consumerConfig.zkConnectionTimeoutMs, ZKStringSerializer)
   // 7. 創建線程池來處理消息流，池的大小是固定的，為partition 的總數，並指定線程池中每一個線程的name 的前綴，內部使用ThreadPoolExecutor，並且 創建線程的 factory類是guava 工具包提供的。
  messageHandlerThreadPool = ThreadUtils.newDaemonFixedThreadPool(
    topics.values.sum, "KafkaMessageHandler")
   // 8. 啟動 BlockGenerator內的兩個線程
  blockGenerator.start()

// 9. 創建MessageStream對象
  val keyDecoder = classTag[U].runtimeClass.getConstructor(classOf[VerifiableProperties])
    .newInstance(consumerConfig.props)
    .asInstanceOf[Decoder[K]]

  val valueDecoder = classTag[T].runtimeClass.getConstructor(classOf[VerifiableProperties])
    .newInstance(consumerConfig.props)
    .asInstanceOf[Decoder[V]]
 
  val topicMessageStreams = consumerConnector.createMessageStreams(
    topics, keyDecoder, valueDecoder)
// 10. 將待處理的MessageHandler 放入 線程池中，等待執行
  topicMessageStreams.values.foreach { streams =>
    streams.foreach { stream =>
      messageHandlerThreadPool.submit(new MessageHandler(stream))
    }
  }
}

其中, 第9 步，創建MessageStream對象，
kafka.consumer.ZookeeperConsumerConnector#createMessageStreams 方法如下：

def createMessageStreams[K,V](topicCountMap: Map[String,Int], keyDecoder: Decoder[K], valueDecoder: Decoder[V])
    : Map[String, List[KafkaStream[K,V]]] = {
  if (messageStreamCreated.getAndSet(true))
    throw new MessageStreamsExistException(this.getClass.getSimpleName +
                                 " can create message streams at most once",null)
  consume(topicCountMap, keyDecoder, valueDecoder)
}

 

其調用了 consume 方法，源碼如下：

def consume[K, V](topicCountMap: scala.collection.Map[String,Int], keyDecoder: Decoder[K], valueDecoder: Decoder[V])
    : Map[String,List[KafkaStream[K,V]]] = {
  debug("entering consume ")
  if (topicCountMap == null)
    throw new RuntimeException("topicCountMap is null")
 // 1. 初始化 topicCount
  val topicCount = TopicCount.constructTopicCount(consumerIdString, topicCountMap)
 // 2. 獲取 每一個topic 和 threadId 集合的映射關系
  val topicThreadIds = topicCount.getConsumerThreadIdsPerTopic

  // make a list of (queue,stream) pairs, one pair for each threadId
// 3. 得到每一個 threadId 對應 (queue, stream) 的映射列表
  val queuesAndStreams = topicThreadIds.values.map(threadIdSet =>
    threadIdSet.map(_ => {
      val queue =  new LinkedBlockingQueue[FetchedDataChunk](config.queuedMaxMessages)
      val stream = new KafkaStream[K,V](
        queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.clientId)
      (queue, stream)
    })
  ).flatten.toList
 // 4. 獲取 groupId 在 zookeeper 中的path
  val dirs = new ZKGroupDirs(config.groupId)
// 5. 注冊 consumer 到 groupId（在zk中）
  registerConsumerInZK(dirs, consumerIdString, topicCount)
// 6. 重新初始化 consumer
  reinitializeConsumer(topicCount, queuesAndStreams)
  // 7. 返回流 
  loadBalancerListener.kafkaMessageAndMetadataStreams.asInstanceOf[Map[String, List[KafkaStream[K,V]]]]
}

 

consumer消費kafka數據

在 kafka.consumer.ZookeeperConsumerConnector#consume方法中，有如下操作：

// 得到每一個 threadId 對應 (queue, stream) 的映射列表
  val queuesAndStreams = topicThreadIds.values.map(threadIdSet =>
    threadIdSet.map(_ => {
      val queue =  new LinkedBlockingQueue[FetchedDataChunk](config.queuedMaxMessages)
      val stream = new KafkaStream[K,V](
        queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.clientId)
      (queue, stream)
    })
  ).flatten.toList
 // 獲取 groupId 在 zookeeper 中的path
  val dirs = new ZKGroupDirs(config.groupId)
// 注冊 consumer 到 groupId（在zk中）
  registerConsumerInZK(dirs, consumerIdString, topicCount)
// 重新初始化 consumer
  reinitializeConsumer(topicCount, queuesAndStreams)

在上面的代碼中，可以看到初始化的queue(LinkedBlockingQueue實例)除了被傳入stream(KafkaStream)的構造函數被迭代器從中取數據，還和 stream 重組成Tuple2[LinkedBlockingQueue[FetchedDataChunk]的list,之后被傳入reinitializeConsumer 方法中。
kafka.consumer.ZookeeperConsumerConnector#reinitializeConsume 其源碼如下：

private def reinitializeConsumer[K,V](
    topicCount: TopicCount,
    queuesAndStreams: List[(LinkedBlockingQueue[FetchedDataChunk],KafkaStream[K,V])]) {
 // 1. 獲取 該groupid 在 zk 中的路徑
  val dirs = new ZKGroupDirs(config.groupId)

  // listener to consumer and partition changes
// 2. 初始化loadBalancerListener，這個負載均衡listener 會時刻監控 consumer 和 partition 的變化
  if (loadBalancerListener == null) {
    val topicStreamsMap = new mutable.HashMap[String,List[KafkaStream[K,V]]]
    loadBalancerListener = new ZKRebalancerListener(
      config.groupId, consumerIdString, topicStreamsMap.asInstanceOf[scala.collection.mutable.Map[String, List[KafkaStream[_,_]]]])
  }

  // create listener for session expired event if not exist yet
  // 3. 監控 session 過期的listner， 有新session注冊初始化，會通知 loadBalancer
if (sessionExpirationListener == null)
    sessionExpirationListener = new ZKSessionExpireListener(
      dirs, consumerIdString, topicCount, loadBalancerListener)

  // create listener for topic partition change event if not exist yet
// 4. 初始化ZKTopicPartitionChangeListener實例，當topic partition 變化時，這個listener會通知 loadBalancer
  if (topicPartitionChangeListener == null)
    topicPartitionChangeListener = new ZKTopicPartitionChangeListener(loadBalancerListener)
 // 5. 將queuesAndStreams 的值經過一系列轉換，並添加到loadBalancerListener.kafkaMessageAndMetadataStreams 中
  val topicStreamsMap = loadBalancerListener.kafkaMessageAndMetadataStreams

  // map of {topic -> Set(thread-1, thread-2, ...)}
  val consumerThreadIdsPerTopic: Map[String, Set[ConsumerThreadId]] =
    topicCount.getConsumerThreadIdsPerTopic

  val allQueuesAndStreams = topicCount match {
    case wildTopicCount: WildcardTopicCount => // 這里是WildcardTopicCount，走這個分支
      /*
       * Wild-card consumption streams share the same queues, so we need to
       * duplicate the list for the subsequent zip operation.
       */
      (1 to consumerThreadIdsPerTopic.keySet.size).flatMap(_ => queuesAndStreams).toList
    case statTopicCount: StaticTopicCount =>
      queuesAndStreams
  }

  val topicThreadIds = consumerThreadIdsPerTopic.map {
    case(topic, threadIds) =>
      threadIds.map((topic, _))
  }.flatten

  require(topicThreadIds.size == allQueuesAndStreams.size,
    "Mismatch between thread ID count (%d) and queue count (%d)"
    .format(topicThreadIds.size, allQueuesAndStreams.size))
  val threadQueueStreamPairs = topicThreadIds.zip(allQueuesAndStreams)

  threadQueueStreamPairs.foreach(e => {
    val topicThreadId = e._1
    val q = e._2._1
    topicThreadIdAndQueues.put(topicThreadId, q)
    debug("Adding topicThreadId %s and queue %s to topicThreadIdAndQueues data structure".format(topicThreadId, q.toString))
    newGauge(
      "FetchQueueSize",
      new Gauge[Int] {
        def value = q.size
      },
      Map("clientId" -> config.clientId,
        "topic" -> topicThreadId._1,
        "threadId" -> topicThreadId._2.threadId.toString)
    )
  })

  val groupedByTopic = threadQueueStreamPairs.groupBy(_._1._1)
  groupedByTopic.foreach(e => {
    val topic = e._1
    val streams = e._2.map(_._2._2).toList
    topicStreamsMap += (topic -> streams)
    debug("adding topic %s and %d streams to map.".format(topic, streams.size))
  })

  // listener to consumer and partition changes
// 6. 使用 zkClient 注冊sessionExpirationListener 實例
  zkClient.subscribeStateChanges(sessionExpirationListener)
 // 7. 使用 zkClient 注冊loadBalancerListener 實例
  zkClient.subscribeChildChanges(dirs.consumerRegistryDir, loadBalancerListener)
 // 遍歷每一個topic，使用zkClient 注冊topicPartitionChangeListener 實例
  topicStreamsMap.foreach { topicAndStreams =>
    // register on broker partition path changes
    val topicPath = BrokerTopicsPath + "/" + topicAndStreams._1
    zkClient.subscribeDataChanges(topicPath, topicPartitionChangeListener)
  }

  // explicitly trigger load balancing for this consumer
// 8. 使用 loadBalancerListener 同步做負載均衡
  loadBalancerListener.syncedRebalance()
}

重點看 第 8 步，使用 loadBalancerListener 同步做負載均衡。
kafka.consumer.ZookeeperConsumerConnector.ZKRebalancerListener#syncedRebalance 源碼如下：

def syncedRebalance() {
  rebalanceLock synchronized {
    rebalanceTimer.time {
      if(isShuttingDown.get())  { // 如果ZookeeperConsumerConnector
已經shutdown了，直接返回
        return
      } else {
        for (i <- 0 until config.rebalanceMaxRetries) { // 默認是 4 次
          info("begin rebalancing consumer " + consumerIdString + " try #" + i)
          var done = false
          var cluster: Cluster = null
          try {
            // 1. 根據zkClient 實例 獲取並創建Cluster 對象，這個 cluster 實例包含了一個 Broker（broker的id，broker在zk中的路徑） 列表
            cluster = getCluster(zkClient) 
            // 2. 在cluster中做 rebalance操作
            done = rebalance(cluster)
          } catch {
            case e: Throwable =>
              /** occasionally, we may hit a ZK exception because the ZK state is changing while we are iterating.
                * For example, a ZK node can disappear between the time we get all children and the time we try to get
                * the value of a child. Just let this go since another rebalance will be triggered.
                **/
              info("exception during rebalance ", e)
          }
          info("end rebalancing consumer " + consumerIdString + " try #" + i)
          if (done) {
            return
          } else {
            /* Here the cache is at a risk of being stale. To take future rebalancing decisions correctly, we should
             * clear the cache */
            info("Rebalancing attempt failed. Clearing the cache before the next rebalancing operation is triggered")
          }
          // stop all fetchers and clear all the queues to avoid data duplication
          closeFetchersForQueues(cluster, kafkaMessageAndMetadataStreams, topicThreadIdAndQueues.map(q => q._2))
          Thread.sleep(config.rebalanceBackoffMs)
        }
      }
    }
  }

  throw new ConsumerRebalanceFailedException(consumerIdString + " can't rebalance after " + config.rebalanceMaxRetries +" retries")
}

 

重點看 第2 步，在 cluster 中做 rebalance 操作，kafka.consumer.ZookeeperConsumerConnector.ZKRebalancerListener#rebalance 源碼如下：

private def rebalance(cluster: Cluster): Boolean = {
  // 1. 獲取 group和 threadId 的Map 映射關系
  val myTopicThreadIdsMap = TopicCount.constructTopicCount(
    group, consumerIdString, zkClient, config.excludeInternalTopics).getConsumerThreadIdsPerTopic
  // 2. 獲取kafka cluster 中所有可用的node
  val brokers = getAllBrokersInCluster(zkClient)
  if (brokers.size == 0) { // 如果可用節點為空，設置listener訂閱，返回 true
    // This can happen in a rare case when there are no brokers available in the cluster when the consumer is started.
    // We log an warning and register for child changes on brokers/id so that rebalance can be triggered when the brokers
    // are up.
    warn("no brokers found when trying to rebalance.")
    zkClient.subscribeChildChanges(ZkUtils.BrokerIdsPath, loadBalancerListener)
    true
  }
  else {
    /**
     * fetchers must be stopped to avoid data duplication, since if the current
     * rebalancing attempt fails, the partitions that are released could be owned by another consumer.
     * But if we don't stop the fetchers first, this consumer would continue returning data for released
     * partitions in parallel. So, not stopping the fetchers leads to duplicate data.
     */
   // 3. 做rebalance 之前的准備工作
   // 3.1. 關閉現有 fetcher 連接
    closeFetchers(cluster, kafkaMessageAndMetadataStreams, myTopicThreadIdsMap)
   // 3.2 釋放 partition 的所有權（主要是刪除zk下的owner 節點的數據以及解除內存中的topic和 fetcher的關聯關系）
    releasePartitionOwnership(topicRegistry)
   // 3.3. 重新給partition分配 fetcher
    val assignmentContext = new AssignmentContext(group, consumerIdString, config.excludeInternalTopics, zkClient)
    val partitionOwnershipDecision = partitionAssignor.assign(assignmentContext)
    val currentTopicRegistry = new Pool[String, Pool[Int, PartitionTopicInfo]](
      valueFactory = Some((topic: String) => new Pool[Int, PartitionTopicInfo]))

    // fetch current offsets for all topic-partitions
    // 3.4 獲取當前fetcher對應的 partitions 的 offsets，這里的offset是指 consumer 下一個要消費的offset
    val topicPartitions = partitionOwnershipDecision.keySet.toSeq

    val offsetFetchResponseOpt = fetchOffsets(topicPartitions)

    if (isShuttingDown.get || !offsetFetchResponseOpt.isDefined)
      false
    else {
      // 3.5 更新 partition 和 fetcher 的對應關系
      val offsetFetchResponse = offsetFetchResponseOpt.get
      topicPartitions.foreach(topicAndPartition => {
        val (topic, partition) = topicAndPartition.asTuple
// requestInfo是OffsetFetchResponse實例中的成員變量，它是一個Map[TopicAndPartition, OffsetMetadataAndError]實例
        val offset = offsetFetchResponse.requestInfo(topicAndPartition).offset
        val threadId = partitionOwnershipDecision(topicAndPartition)
        addPartitionTopicInfo(currentTopicRegistry, partition, topic, offset, threadId)
      })

      /**
       * move the partition ownership here, since that can be used to indicate a truly successful rebalancing attempt
       * A rebalancing attempt is completed successfully only after the fetchers have been started correctly
       */
      if(reflectPartitionOwnershipDecision(partitionOwnershipDecision)) {
        allTopicsOwnedPartitionsCount = partitionOwnershipDecision.size

        partitionOwnershipDecision.view.groupBy { case(topicPartition, consumerThreadId) => topicPartition.topic }
                                  .foreach { case (topic, partitionThreadPairs) =>
          newGauge("OwnedPartitionsCount",
            new Gauge[Int] {
              def value() = partitionThreadPairs.size
            },
            ownedPartitionsCountMetricTags(topic))
        }
        // 3.6 將已經新的 topic registry 覆蓋舊的
        topicRegistry = currentTopicRegistry
// 4. 更新 fetcher
        updateFetcher(cluster)
        true
      } else {
        false
      }
    }
  }
}

其中addPartitionTopicInfo 源碼如下：

private def addPartitionTopicInfo(currentTopicRegistry: Pool[String, Pool[Int, PartitionTopicInfo]],
                                    partition: Int, topic: String,
                                    offset: Long, consumerThreadId: ConsumerThreadId) {
//如果map沒有對應的 key，會使用valueFactory初始化鍵值對，並返回 對應的 value
    val partTopicInfoMap = currentTopicRegistry.getAndMaybePut(topic)

    val queue = topicThreadIdAndQueues.get((topic, consumerThreadId))
    val consumedOffset = new AtomicLong(offset)
    val fetchedOffset = new AtomicLong(offset)
    val partTopicInfo = new PartitionTopicInfo(topic,
                                               partition,
                                               queue,
                                               consumedOffset,
                                               fetchedOffset,
                                               new AtomicInteger(config.fetchMessageMaxBytes),
                                               config.clientId)
    // 1. 將其注冊到新的 Topic注冊中心中，即注冊 partition 和 fetcher 的關系
partTopicInfoMap.put(partition, partTopicInfo)
    debug(partTopicInfo + " selected new offset " + offset)
// 2. 更新consumer 的 已經消費的offset信息
    checkpointedZkOffsets.put(TopicAndPartition(topic, partition), offset)
  }
}

 

第4步， 更新 fetcher 源碼如下：

private def updateFetcher(cluster: Cluster) {
  // update partitions for fetcher
  var allPartitionInfos : List[PartitionTopicInfo] = Nil
  for (partitionInfos <- topicRegistry.values)
    for (partition <- partitionInfos.values)
      allPartitionInfos ::= partition
  info("Consumer " + consumerIdString + " selected partitions : " +
    allPartitionInfos.sortWith((s,t) => s.partitionId < t.partitionId).map(_.toString).mkString(","))

  fetcher match {
    case Some(f) =>
      f.startConnections(allPartitionInfos, cluster)
    case None =>
  }
}

 

其中，f.startConnections方法真正執行 更新操作。此時引入一個新的類。即 fetcher 類，kafka.consumer.ConsumerFetcherManager。

kafka.consumer.ConsumerFetcherManager#startConnections 的源碼如下：

def startConnections(topicInfos: Iterable[PartitionTopicInfo], cluster: Cluster) {
// LeaderFinderThread 在 topic 的leader node可用時，將 fetcher 添加到 leader 節點上
  leaderFinderThread = new LeaderFinderThread(consumerIdString + "-leader-finder-thread")
  leaderFinderThread.start()

  inLock(lock) {
// 更新ConsumerFetcherManager 成員變量
    partitionMap = topicInfos.map(tpi => (TopicAndPartition(tpi.topic, tpi.partitionId), tpi)).toMap
    this.cluster = cluster
    noLeaderPartitionSet ++= topicInfos.map(tpi => TopicAndPartition(tpi.topic, tpi.partitionId))
    cond.signalAll()
  }
}

 

ConsumerFetcherManager 有一個LeaderFinderThread 實例，該類的父類kafka.utils.ShutdownableThread ，run 方法如下：

override def run(): Unit = {
  info("Starting ")
  try{
    while(isRunning.get()){
      doWork()
    }
  } catch{
    case e: Throwable =>
      if(isRunning.get())
        error("Error due to ", e)
  }
  shutdownLatch.countDown()
  info("Stopped ")
}

doWork其實就是一個抽象方法，其子類LeaderFinderThread的實現如下：

// thread responsible for adding the fetcher to the right broker when leader is available
override def doWork() {
// 1. 獲取 partition 和leader node的映射關系
  val leaderForPartitionsMap = new HashMap[TopicAndPartition, Broker]
  lock.lock()
  try {
    while (noLeaderPartitionSet.isEmpty) { // 這個字段在startConnections 已更新新值
      trace("No partition for leader election.")
      cond.await()
    }

    trace("Partitions without leader %s".format(noLeaderPartitionSet))
    val brokers = getAllBrokersInCluster(zkClient) // 獲取所有可用broker 節點
    // 獲取kafka.api.TopicMetadata 序列，kafka.api.TopicMetadata 保存了 topic 和 partitionId，isr，leader，replicas 的信息
val topicsMetadata = ClientUtils.fetchTopicMetadata(noLeaderPartitionSet.map(m => m.topic).toSet,
                                                        brokers,
                                                        config.clientId,
                                                        config.socketTimeoutMs,
                                                        correlationId.getAndIncrement).topicsMetadata
    if(logger.isDebugEnabled) topicsMetadata.foreach(topicMetadata => debug(topicMetadata.toString()))
// 2. 根據獲取到的 partition 和 leader node 的關系更新noLeaderPartitionSet 和leaderForPartitionsMap 兩個map集合，其中noLeaderPartitionSet 包含的是沒有確定leader 的 partition 集合，leaderForPartitionsMap 是 已經確定了 leader 的 partition 集合。
    topicsMetadata.foreach { tmd =>
      val topic = tmd.topic
      tmd.partitionsMetadata.foreach { pmd =>
        val topicAndPartition = TopicAndPartition(topic, pmd.partitionId)
        if(pmd.leader.isDefined && noLeaderPartitionSet.contains(topicAndPartition)) {
          val leaderBroker = pmd.leader.get
          leaderForPartitionsMap.put(topicAndPartition, leaderBroker)
          noLeaderPartitionSet -= topicAndPartition
        }
      }
    }
  } catch {
    case t: Throwable => {
        if (!isRunning.get())
          throw t /* If this thread is stopped, propagate this exception to kill the thread. */
        else
          warn("Failed to find leader for %s".format(noLeaderPartitionSet), t)
      }
  } finally {
    lock.unlock()
  }

  try {
// 3. 具體為 partition 分配 fetcher
    addFetcherForPartitions(leaderForPartitionsMap.map{
      case (topicAndPartition, broker) =>
        topicAndPartition -> BrokerAndInitialOffset(broker, partitionMap(topicAndPartition).getFetchOffset())}
    )
  } catch {
    case t: Throwable => {
      if (!isRunning.get())
        throw t /* If this thread is stopped, propagate this exception to kill the thread. */
      else {
        warn("Failed to add leader for partitions %s; will retry".format(leaderForPartitionsMap.keySet.mkString(",")), t)
        lock.lock()
        noLeaderPartitionSet ++= leaderForPartitionsMap.keySet
        lock.unlock()
      }
    }
  }
  // 4. 關閉空閑fetcher線程
  shutdownIdleFetcherThreads()
  Thread.sleep(config.refreshLeaderBackoffMs)
}

 

重點看第3 步，具體為 partition 分配 fetcher，addFetcherForPartitions 源碼如下：

def addFetcherForPartitions(partitionAndOffsets: Map[TopicAndPartition, BrokerAndInitialOffset]) {
  mapLock synchronized {
// 獲取 fetcher 和 partition的映射關系
    val partitionsPerFetcher = partitionAndOffsets.groupBy{ case(topicAndPartition, brokerAndInitialOffset) =>
      BrokerAndFetcherId(brokerAndInitialOffset.broker, getFetcherId(topicAndPartition.topic, topicAndPartition.partition))}
    for ((brokerAndFetcherId, partitionAndOffsets) <- partitionsPerFetcher) {

      var fetcherThread: AbstractFetcherThread = null
      fetcherThreadMap.get(brokerAndFetcherId) match {
        case Some(f) => fetcherThread = f
        case None =>
// 根據brokerAndFetcherId 去初始化Fetcher並啟動 fetcher
          fetcherThread = createFetcherThread(brokerAndFetcherId.fetcherId, brokerAndFetcherId.broker)
          fetcherThreadMap.put(brokerAndFetcherId, fetcherThread)
          fetcherThread.start
      }

      fetcherThreadMap(brokerAndFetcherId).addPartitions(partitionAndOffsets.map { case (topicAndPartition, brokerAndInitOffset) =>
        topicAndPartition -> brokerAndInitOffset.initOffset
      })
    }
  }

  info("Added fetcher for partitions %s".format(partitionAndOffsets.map{ case (topicAndPartition, brokerAndInitialOffset) =>
    "[" + topicAndPartition + ", initOffset " + brokerAndInitialOffset.initOffset + " to broker " + brokerAndInitialOffset.broker + "] "}))
}

 

kafka.consumer.ConsumerFetcherManager#createFetcherThread的源碼如下：

override def createFetcherThread(fetcherId: Int, sourceBroker: Broker): AbstractFetcherThread = {
  new ConsumerFetcherThread(
    "ConsumerFetcherThread-%s-%d-%d".format(consumerIdString, fetcherId, sourceBroker.id),
    config, sourceBroker, partitionMap, this)
}

 

先來看ConsumerFetcherThread的構造方法聲明：

class ConsumerFetcherThread(name: String,
                            val config: ConsumerConfig,
                            sourceBroker: Broker,
                            partitionMap: Map[TopicAndPartition, PartitionTopicInfo],
                            val consumerFetcherManager: ConsumerFetcherManager)
        extends AbstractFetcherThread(name = name, 
                                      clientId = config.clientId,
                                      sourceBroker = sourceBroker,
                                      socketTimeout = config.socketTimeoutMs,
                                      socketBufferSize = config.socketReceiveBufferBytes,
                                      fetchSize = config.fetchMessageMaxBytes,
                                      fetcherBrokerId = Request.OrdinaryConsumerId,
                                      maxWait = config.fetchWaitMaxMs,
                                      minBytes = config.fetchMinBytes,
                                      isInterruptible = true)

注意，partitionMap 中的value 是PartitionTopicInfo ，這個對象中封裝了存放fetch結果值的BlockingQueue[FetchedDataChunk] 實例。
再來看 run 方法，其使用的是 kafka.utils.ShutdownableThread#run 方法，上面我們已經看過了，主要看該子類是如何重新 doWork方法的：

override def doWork() {
  inLock(partitionMapLock) { // 加鎖，執行，釋放鎖
    if (partitionMap.isEmpty) // 如果沒有需要執行的 fetch 操作，等待200ms后返回
      partitionMapCond.await(200L, TimeUnit.MILLISECONDS)
    partitionMap.foreach { // 將所有的 fetch 的信息添加到fetchRequestBuilder中
      case((topicAndPartition, offset)) =>
        fetchRequestBuilder.addFetch(topicAndPartition.topic, topicAndPartition.partition,
                         offset, fetchSize)
    }
  }
  // 構建批抓取的fetchRequest對象
  val fetchRequest = fetchRequestBuilder.build()
// 處理 FetchRequest
  if (!fetchRequest.requestInfo.isEmpty)
    processFetchRequest(fetchRequest)
}

 

其中 kafka.server.AbstractFetcherThread#processFetchRequest 源碼如下：

private def processFetchRequest(fetchRequest: FetchRequest) {
  val partitionsWithError = new mutable.HashSet[TopicAndPartition]
  var response: FetchResponse = null
  try {
    trace("Issuing to broker %d of fetch request %s".format(sourceBroker.id, fetchRequest))
// 發送請求，並獲取返回值。
// simpleConsumer  就是SimpleConsumer 實例，已作說明，不再贅述。
    response = simpleConsumer.fetch(fetchRequest)
  } catch {
    case t: Throwable =>
      if (isRunning.get) {
        warn("Error in fetch %s. Possible cause: %s".format(fetchRequest, t.toString))
        partitionMapLock synchronized {
          partitionsWithError ++= partitionMap.keys
        }
      }
  }
  fetcherStats.requestRate.mark()

  if (response != null) {
    // process fetched data
    inLock(partitionMapLock) { // 獲取鎖，執行處理response 操作，釋放鎖
      response.data.foreach {
        case(topicAndPartition, partitionData) =>
          val (topic, partitionId) = topicAndPartition.asTuple
          val currentOffset = partitionMap.get(topicAndPartition)
          // we append to the log if the current offset is defined and it is the same as the offset requested during fetch
          if (currentOffset.isDefined && fetchRequest.requestInfo(topicAndPartition).offset == currentOffset.get) {
            partitionData.error match { // 根據返回碼來確定具體執行哪部分處理邏輯
              case ErrorMapping.NoError => // 成功返回，沒有錯誤
                try {
                  val messages = partitionData.messages.asInstanceOf[ByteBufferMessageSet]
                  val validBytes = messages.validBytes
                  val newOffset = messages.shallowIterator.toSeq.lastOption match {
                    case Some(m: MessageAndOffset) => m.nextOffset
                    case None => currentOffset.get
                  }
                  partitionMap.put(topicAndPartition, newOffset)
                  fetcherLagStats.getFetcherLagStats(topic, partitionId).lag = partitionData.hw - newOffset
                  fetcherStats.byteRate.mark(validBytes)
                  // Once we hand off the partition data to the subclass, we can't mess with it any more in this thread
                  processPartitionData(topicAndPartition, currentOffset.get, partitionData)
                } catch {
                  case ime: InvalidMessageException => // 消息獲取不完整
                    // we log the error and continue. This ensures two things
                    // 1. If there is a corrupt message in a topic partition, it does not bring the fetcher thread down and cause other topic partition to also lag
                    // 2. If the message is corrupt due to a transient state in the log (truncation, partial writes can cause this), we simply continue and
                    //    should get fixed in the subsequent fetches
                    logger.error("Found invalid messages during fetch for partition [" + topic + "," + partitionId + "] offset " + currentOffset.get + " error " + ime.getMessage)
                  case e: Throwable =>
                    throw new KafkaException("error processing data for partition [%s,%d] offset %d"
                                             .format(topic, partitionId, currentOffset.get), e)
                }
              case ErrorMapping.OffsetOutOfRangeCode => // offset out of range error
                try {
                  val newOffset = handleOffsetOutOfRange(topicAndPartition)
                  partitionMap.put(topicAndPartition, newOffset)
                  error("Current offset %d for partition [%s,%d] out of range; reset offset to %d"
                    .format(currentOffset.get, topic, partitionId, newOffset))
                } catch {
                  case e: Throwable =>
                    error("Error getting offset for partition [%s,%d] to broker %d".format(topic, partitionId, sourceBroker.id), e)
                    partitionsWithError += topicAndPartition
                }
              case _ =>
                if (isRunning.get) {
                  error("Error for partition [%s,%d] to broker %d:%s".format(topic, partitionId, sourceBroker.id,
                    ErrorMapping.exceptionFor(partitionData.error).getClass))
                  partitionsWithError += topicAndPartition
                }
            }
          }
      }
    }
  }

  if(partitionsWithError.size > 0) {
    debug("handling partitions with error for %s".format(partitionsWithError))
    handlePartitionsWithErrors(partitionsWithError)
  }
}

 

其中processPartitionData 源碼如下，它負責處理具體的返回消息：

// process fetched data
def processPartitionData(topicAndPartition: TopicAndPartition, fetchOffset: Long, partitionData: FetchResponsePartitionData) {
// partitionMap 是一個成員變量，在構造函數中作為入參
  val pti = partitionMap(topicAndPartition)
  if (pti.getFetchOffset != fetchOffset)
    throw new RuntimeException("Offset doesn't match for partition [%s,%d] pti offset: %d fetch offset: %d"
                              .format(topicAndPartition.topic, topicAndPartition.partition, pti.getFetchOffset, fetchOffset))
// 數據入隊
  pti.enqueue(partitionData.messages.asInstanceOf[ByteBufferMessageSet])
}

可以看到，終於在這里，把從leader中fetch的消息放入了BlockingQueue[FetchedDataChunk] 緩沖堵塞隊列中。

KafkaStream從queue中堵塞式獲取數據

KafkaStream 是依賴於 LinkedBlockingQueue 的同理 KafkaStream 也會返回一個迭代器 kafka.consumer.ConsumerIterator，用於迭代訪問 KafkaStream 中的數據。
kafka.consumer.ConsumerIterator 的主要源碼如下：

// 判斷是否有下一個元素
def hasNext(): Boolean = {
  if(state == FAILED)
    throw new IllegalStateException("Iterator is in failed state")
  state match {
    case DONE => false
    case READY => true
    case _ => maybeComputeNext()
  }
}
// 獲取下一個元素，父類實現
def next(): T = {
  if(!hasNext())
    throw new NoSuchElementException()
  state = NOT_READY
  if(nextItem == null)
    throw new IllegalStateException("Expected item but none found.")
  nextItem
}
// 獲取下一個元素，使用子類ConsumerIterator實現
override def next(): MessageAndMetadata[K, V] = {
  val item = super.next() // 調用父類實現
  if(consumedOffset < 0)
    throw new KafkaException("Offset returned by the message set is invalid %d".format(consumedOffset))
  currentTopicInfo.resetConsumeOffset(consumedOffset)
  val topic = currentTopicInfo.topic
  trace("Setting %s consumed offset to %d".format(topic, consumedOffset))
  consumerTopicStats.getConsumerTopicStats(topic).messageRate.mark()
  consumerTopicStats.getConsumerAllTopicStats().messageRate.mark()
  item
}
 // 或許有，嘗試計算一下下一個
def maybeComputeNext(): Boolean = {
  state = FAILED
  nextItem = makeNext()
  if(state == DONE) {
    false
  } else {
    state = READY
    true
  }
}
// 創建下一個元素，這個在子類ConsumerIterator中有實現
protected def makeNext(): MessageAndMetadata[K, V] = {
// 首先channel 是 LinkedBlockingQueue實例， 是 KafkaStream 中的 queue 成員變量，queue 成員變量
  var currentDataChunk: FetchedDataChunk = null
  // if we don't have an iterator, get one
  var localCurrent = current.get() 
// 如果沒有迭代器或者是沒有下一個元素了，需要從channel中取一個
  if(localCurrent == null || !localCurrent.hasNext) {
// 刪除並返回隊列的頭節點。
    if (consumerTimeoutMs < 0)
      currentDataChunk = channel.take // 阻塞方法，一直等待，直到有可用元素
    else {
      currentDataChunk = channel.poll(consumerTimeoutMs,  TimeUnit.MILLISECONDS) // 阻塞方法，等待指定時間，超時也會返回
      if (currentDataChunk == null) { // 如果沒有數據，重置狀態為NOT_READY
        // reset state to make the iterator re-iterable
        resetState()
        throw new ConsumerTimeoutException
      }
    }
// 關閉命令
    if(currentDataChunk eq ZookeeperConsumerConnector.shutdownCommand) {
      debug("Received the shutdown command")
      return allDone // 該函數將狀態設為DONE， 返回null
    } else {
      currentTopicInfo = currentDataChunk.topicInfo
      val cdcFetchOffset = currentDataChunk.fetchOffset
      val ctiConsumeOffset = currentTopicInfo.getConsumeOffset
      if (ctiConsumeOffset < cdcFetchOffset) {
        error("consumed offset: %d doesn't match fetch offset: %d for %s;\n Consumer may lose data"
          .format(ctiConsumeOffset, cdcFetchOffset, currentTopicInfo))
        currentTopicInfo.resetConsumeOffset(cdcFetchOffset)
      }
      localCurrent = currentDataChunk.messages.iterator

      current.set(localCurrent)
    }
    // if we just updated the current chunk and it is empty that means the fetch size is too small!
    if(currentDataChunk.messages.validBytes == 0)
      throw new MessageSizeTooLargeException("Found a message larger than the maximum fetch size of this consumer on topic " +
                                             "%s partition %d at fetch offset %d. Increase the fetch size, or decrease the maximum message size the broker will allow."
                                             .format(currentDataChunk.topicInfo.topic, currentDataChunk.topicInfo.partitionId, currentDataChunk.fetchOffset))
  }
  var item = localCurrent.next()
  // reject the messages that have already been consumed
  while (item.offset < currentTopicInfo.getConsumeOffset && localCurrent.hasNext) {
    item = localCurrent.next()
  }
  consumedOffset = item.nextOffset

  item.message.ensureValid() // validate checksum of message to ensure it is valid
 // 返回處理封裝好的 kafka 數據
  new MessageAndMetadata(currentTopicInfo.topic, currentTopicInfo.partitionId, item.message, item.offset, keyDecoder, valueDecoder)
}

 

消費到的數據cache 到WAL中

我們再來看，org.apache.spark.streaming.kafka.ReliableKafkaReceiver#onStart 的第10 步相應的代碼：

// 10. 將待處理的MessageHandler 放入 線程池中，等待執行
  topicMessageStreams.values.foreach { streams =>
    streams.foreach { stream =>
      messageHandlerThreadPool.submit(new MessageHandler(stream))
    }
  }

其中 MessageHandler 是一個 Runnable 對象，其 run 方法如下：

override def run(): Unit = {
  while (!isStopped) {
    try {
// 1. 獲取ConsumerIterator 迭代器對象
      val streamIterator = stream.iterator()
      // 2. 遍歷迭代器中獲取每一條數據，並且保存message和相應的 metadata 信息
while (streamIterator.hasNext) {
        storeMessageAndMetadata(streamIterator.next)
      }
    } catch {
      case e: Exception =>
        reportError("Error handling message", e)
    }
  }
}

 

其中第二步中關鍵方法，org.apache.spark.streaming.kafka.ReliableKafkaReceiver#storeMessageAndMetadata 方法如下：

/** Store a Kafka message and the associated metadata as a tuple. */
private def storeMessageAndMetadata(
    msgAndMetadata: MessageAndMetadata[K, V]): Unit = {
  val topicAndPartition = TopicAndPartition(msgAndMetadata.topic, msgAndMetadata.partition)
  val data = (msgAndMetadata.key, msgAndMetadata.message)
  val metadata = (topicAndPartition, msgAndMetadata.offset)
// 添加數據到 block
  blockGenerator.addDataWithCallback(data, metadata)
}

addDataWithCallback 源碼如下：

/**
 * Push a single data item into the buffer. After buffering the data, the
 * `BlockGeneratorListener.onAddData` callback will be called.
 */
def addDataWithCallback(data: Any, metadata: Any): Unit = {
  if (state == Active) {
    waitToPush()
    synchronized {
      if (state == Active) {
// 1. 將數據放入 buffer 中，以便處理線程從中獲取數據
        currentBuffer += data
// 2. 在啟動 receiver線程中，可以知道listener 是指GeneratedBlockHandler 實例
        listener.onAddData(data, metadata)
      } else {
        throw new SparkException(
          "Cannot add data as BlockGenerator has not been started or has been stopped")
      }
    }
  } else {
    throw new SparkException(
      "Cannot add data as BlockGenerator has not been started or has been stopped")
  }
}

 

第二步比較簡單，先看一下第二步：
org.apache.spark.streaming.kafka.ReliableKafkaReceiver.GeneratedBlockHandler#onAddData的源碼如下：

def onAddData(data: Any, metadata: Any): Unit = {
  // Update the offset of the data that was added to the generator
  if (metadata != null) {
    val (topicAndPartition, offset) = metadata.asInstanceOf[(TopicAndPartition, Long)]
    updateOffset(topicAndPartition, offset)
  }
}
// 這里的 updateOffset 調用的是//org.apache.spark.streaming.kafka.ReliableKafkaReceiver#updateOffset，源碼如下：
/** Update stored offset */
private def updateOffset(topicAndPartition: TopicAndPartition, offset: Long): Unit = {
  topicPartitionOffsetMap.put(topicAndPartition, offset)
}

 

第一步的原理如下：
在 BlockGenerator中有一個定時器，定時（200ms）去執行檢查currentBuffer是否為empty任務， 若不為空，則執行如下操作並把它放入等待生成block 的隊列中，有兩外一個線程來時刻監聽這個隊列，有數據，則執行pushBlock 操作。
第一個定時器線程如下：

private val blockIntervalTimer =
  new RecurringTimer(clock, blockIntervalMs, updateCurrentBuffer, "BlockGenerator")

// 其中，updateCurrentBuffer 方法如下
/** Change the buffer to which single records are added to. */
private def updateCurrentBuffer(time: Long): Unit = {
  try {
    var newBlock: Block = null
    synchronized {
      if (currentBuffer.nonEmpty) {
        val newBlockBuffer = currentBuffer
        currentBuffer = new ArrayBuffer[Any]
        val blockId = StreamBlockId(receiverId, time - blockIntervalMs)
        listener.onGenerateBlock(blockId)
        newBlock = new Block(blockId, newBlockBuffer)
      }
    }

    if (newBlock != null) {
      blocksForPushing.put(newBlock)  // put is blocking when queue is full
    }
  } catch {
    case ie: InterruptedException =>
      logInfo("Block updating timer thread was interrupted")
    case e: Exception =>
      reportError("Error in block updating thread", e)
  }
}

// listener.onGenerateBlock(blockId) 代碼如下：
def onGenerateBlock(blockId: StreamBlockId): Unit = {
  // Remember the offsets of topics/partitions when a block has been generated
  rememberBlockOffsets(blockId)
}
// rememberBlockOffsets 代碼如下：
/**
 * Remember the current offsets for each topic and partition. This is called when a block is
 * generated.
 */
private def rememberBlockOffsets(blockId: StreamBlockId): Unit = {
  // Get a snapshot of current offset map and store with related block id.
  val offsetSnapshot = topicPartitionOffsetMap.toMap
  blockOffsetMap.put(blockId, offsetSnapshot)
  topicPartitionOffsetMap.clear()
}
// 可以看出，主要是清除 topic-partition-> offset 映射關系
// 建立 block 和topic-partition-> offset的映射關系

其中，blocksForPushing是一個有界阻塞隊列，另外一個線程會一直輪詢它。

private val blocksForPushing = new ArrayBlockingQueue[Block](blockQueueSize)
private val blockPushingThread = new Thread() { override def run() { keepPushingBlocks() } }

/** Keep pushing blocks to the BlockManager. */
// 這個方法主要的作用就是一直不停地輪詢blocksForPushing隊列，並處理相應的push block 事件。
private def keepPushingBlocks() {
  logInfo("Started block pushing thread")

  def areBlocksBeingGenerated: Boolean = synchronized {
    state != StoppedGeneratingBlocks
  }

  try {
    // While blocks are being generated, keep polling for to-be-pushed blocks and push them.
    while (areBlocksBeingGenerated) { // 線程沒有被停止，則一直循環
// 超時poll操作獲取並刪除頭節點，超過時間（10ms）則返回
      Option(blocksForPushing.poll(10, TimeUnit.MILLISECONDS)) match {
        case Some(block) => pushBlock(block) // 如果有數據則進行處理。
        case None =>
      }
    }

    // At this point, state is StoppedGeneratingBlock. So drain the queue of to-be-pushed blocks.
    logInfo("Pushing out the last " + blocksForPushing.size() + " blocks")
    while (!blocksForPushing.isEmpty) { // 如果隊列中還有數據，繼續進行處理
      val block = blocksForPushing.take() // 這是一個堵塞方法，不過現在會馬上返回，因為隊列里面有數據。
      logDebug(s"Pushing block $block")
      pushBlock(block) // 處理數據
      logInfo("Blocks left to push " + blocksForPushing.size())
    }
    logInfo("Stopped block pushing thread")
  } catch {
    case ie: InterruptedException =>
      logInfo("Block pushing thread was interrupted")
    case e: Exception =>
      reportError("Error in block pushing thread", e)
  }
}

 

其中的pushBlock源碼如下：

private def pushBlock(block: Block) {
  listener.onPushBlock(block.id, block.buffer)
  logInfo("Pushed block " + block.id)
}

其調用的listener(org.apache.spark.streaming.kafka.ReliableKafkaReceiver.GeneratedBlockHandler)的 onPushBlock 源碼如下：

def onPushBlock(blockId: StreamBlockId, arrayBuffer: mutable.ArrayBuffer[_]): Unit = {
  // Store block and commit the blocks offset
  storeBlockAndCommitOffset(blockId, arrayBuffer)
}

其中，storeBlockAndCommitOffset具體代碼如下：

/**
 * Store the ready-to-be-stored block and commit the related offsets to zookeeper. This method
 * will try a fixed number of times to push the block. If the push fails, the receiver is stopped.
 */
private def storeBlockAndCommitOffset(
    blockId: StreamBlockId, arrayBuffer: mutable.ArrayBuffer[_]): Unit = {
  var count = 0
  var pushed = false
  var exception: Exception = null
  while (!pushed && count <= 3) { // 整個過程，總共允許3 次重試
    try {
      store(arrayBuffer.asInstanceOf[mutable.ArrayBuffer[(K, V)]])
      pushed = true
    } catch {
      case ex: Exception =>
        count += 1
        exception = ex
    }
  }
  if (pushed) { // 已經push block
// 更新 offset
    Option(blockOffsetMap.get(blockId)).foreach(commitOffset)
// 如果已經push 到 BlockManager 中，則不會再保留 block和topic-partition-> offset的映射關系
    blockOffsetMap.remove(blockId)
  } else {
    stop("Error while storing block into Spark", exception)
  }
}
// 其中，commitOffset源碼如下：
/**
 * Commit the offset of Kafka's topic/partition, the commit mechanism follow Kafka 0.8.x's
 * metadata schema in Zookeeper.
 */
private def commitOffset(offsetMap: Map[TopicAndPartition, Long]): Unit = {
  if (zkClient == null) {
    val thrown = new IllegalStateException("Zookeeper client is unexpectedly null")
    stop("Zookeeper client is not initialized before commit offsets to ZK", thrown)
    return
  }

  for ((topicAndPart, offset) <- offsetMap) {
    try {
// 獲取在 zk 中 comsumer 的partition的目錄
      val topicDirs = new ZKGroupTopicDirs(groupId, topicAndPart.topic)
      val zkPath = s"${topicDirs.consumerOffsetDir}/${topicAndPart.partition}"
      // 更新 consumer 的已消費topic-partition 的offset 操作
      ZkUtils.updatePersistentPath(zkClient, zkPath, offset.toString)
    } catch {
      case e: Exception =>
        logWarning(s"Exception during commit offset $offset for topic" +
          s"${topicAndPart.topic}, partition ${topicAndPart.partition}", e)
    }

    logInfo(s"Committed offset $offset for topic ${topicAndPart.topic}, " +
      s"partition ${topicAndPart.partition}")
  }
}

關鍵方法store 如下：

/** Store an ArrayBuffer of received data as a data block into Spark's memory. */
def store(dataBuffer: ArrayBuffer[T]) {
  supervisor.pushArrayBuffer(dataBuffer, None, None)
}

其調用了supervisor（org.apache.spark.streaming.receiver.ReceiverSupervisorImpl實例）的pushArrayBuffer方法，內部操作如下：

/** Store an ArrayBuffer of received data as a data block into Spark's memory. */
def pushArrayBuffer(
    arrayBuffer: ArrayBuffer[_],
    metadataOption: Option[Any],
    blockIdOption: Option[StreamBlockId]
  ) {
  pushAndReportBlock(ArrayBufferBlock(arrayBuffer), metadataOption, blockIdOption)
}

org.apache.spark.streaming.receiver.ReceiverSupervisorImpl#pushAndReportBlock 源碼如下：

/** Store block and report it to driver */
def pushAndReportBlock(
    receivedBlock: ReceivedBlock,
    metadataOption: Option[Any],
    blockIdOption: Option[StreamBlockId]
  ) {
// 1.准備blockId，time等信息
  val blockId = blockIdOption.getOrElse(nextBlockId)
  val time = System.currentTimeMillis
// 2. 執行存儲 block 操作
  val blockStoreResult = receivedBlockHandler.storeBlock(blockId, receivedBlock)
  logDebug(s"Pushed block $blockId in ${(System.currentTimeMillis - time)} ms")
// 3. 獲取保存的message 的記錄數
  val numRecords = blockStoreResult.numRecords
// 4. 通知trackerEndpoint已經添加block，執行更新driver 的WAL操作
  val blockInfo = ReceivedBlockInfo(streamId, numRecords, metadataOption, blockStoreResult)
  trackerEndpoint.askWithRetry[Boolean](AddBlock(blockInfo))
  logDebug(s"Reported block $blockId")
}

其中，receivedBlockHandler 的賦值語句如下：

private val receivedBlockHandler: ReceivedBlockHandler = {
  if (WriteAheadLogUtils.enableReceiverLog(env.conf)) {
    if (checkpointDirOption.isEmpty) {
      throw new SparkException(
        "Cannot enable receiver write-ahead log without checkpoint directory set. " +
          "Please use streamingContext.checkpoint() to set the checkpoint directory. " +
          "See documentation for more details.")
    }
// enable WAL並且checkpoint dir 不為空，即，在這里，返回WriteAheadLogBasedBlockHandler 對象，這個對象持有了 blockmanager，streamid，storagelevel，conf，checkpointdir 等信息
    new WriteAheadLogBasedBlockHandler(env.blockManager, receiver.streamId,
      receiver.storageLevel, env.conf, hadoopConf, checkpointDirOption.get)
  } else {
    new BlockManagerBasedBlockHandler(env.blockManager, receiver.storageLevel)
  }
}

ReceivedBlockHandler 的 storeBlock方法源碼如下：

/**
 * This implementation stores the block into the block manager as well as a write ahead log.
 * It does this in parallel, using Scala Futures, and returns only after the block has
 * been stored in both places.
 */
// 並行地將block 存入 blockmanager 和 write ahead log，使用scala 的Future 機制實現的，當兩個都寫完畢之后，返回。
def storeBlock(blockId: StreamBlockId, block: ReceivedBlock): ReceivedBlockStoreResult = {

  var numRecords = None: Option[Long]
  // Serialize the block so that it can be inserted into both
// 1. 將ReceivedBlock序列化（未使用壓縮機制）成字節數組
  val serializedBlock = block match { // serializedBlock 就是序列化后的結果
    case ArrayBufferBlock(arrayBuffer) => // go this branch
      numRecords = Some(arrayBuffer.size.toLong)
      blockManager.dataSerialize(blockId, arrayBuffer.iterator)
    case IteratorBlock(iterator) =>
      val countIterator = new CountingIterator(iterator)
      val serializedBlock = blockManager.dataSerialize(blockId, countIterator)
      numRecords = countIterator.count
      serializedBlock
    case ByteBufferBlock(byteBuffer) =>
      byteBuffer
    case _ =>
      throw new Exception(s"Could not push $blockId to block manager, unexpected block type")
  }

  // 2. Store the block in block manager
  val storeInBlockManagerFuture = Future {
    val putResult =
      blockManager.putBytes(blockId, serializedBlock, effectiveStorageLevel, tellMaster = true)
    if (!putResult.map { _._1 }.contains(blockId)) {
      throw new SparkException(
        s"Could not store $blockId to block manager with storage level $storageLevel")
    }
  }

  // 3. Store the block in write ahead log
  val storeInWriteAheadLogFuture = Future {
    writeAheadLog.write(serializedBlock, clock.getTimeMillis())
  }

  // 4. Combine the futures, wait for both to complete, and return the write ahead log record handle
  val combinedFuture = storeInBlockManagerFuture.zip(storeInWriteAheadLogFuture).map(_._2)
// 等待future任務結果返回。默認時間是 30s， 使用spark.streaming.receiver.blockStoreTimeout 參數來變更默認值
  val walRecordHandle = Await.result(combinedFuture, blockStoreTimeout)
  // 返回cache之后的block 相關信息
WriteAheadLogBasedStoreResult(blockId, numRecords, walRecordHandle)
}

將WAL的block信息發送給driver

注意WriteAheadLogBasedStoreResult 這個 WriteAheadLogBasedStoreResult 實例，后面 RDD 在處理的時候會使用到。
org.apache.spark.streaming.receiver.ReceiverSupervisorImpl#pushAndReportBlock 通知driver addBlock 的源碼如下：

// 4. 通知trackerEndpoint已經添加block，執行更新driver 的WAL操作
  val blockInfo = ReceivedBlockInfo(streamId, numRecords, metadataOption, blockStoreResult)
  trackerEndpoint.askWithRetry[Boolean](AddBlock(blockInfo))
  logDebug(s"Reported block $blockId")

Driver將WAL block數據寫入到 driver 的WAL中

跳過中間的RPC操作，直接到 driver 端org.apache.spark.streaming.scheduler.ReceiverTracker.ReceiverTrackerEndpoint#receiveAndReply 中：

case AddBlock(receivedBlockInfo) =>
  if (WriteAheadLogUtils.isBatchingEnabled(ssc.conf, isDriver = true)) {
    walBatchingThreadPool.execute(new Runnable {
      override def run(): Unit = Utils.tryLogNonFatalError {
        if (active) {
          context.reply(addBlock(receivedBlockInfo))
        } else {
          throw new IllegalStateException("ReceiverTracker RpcEndpoint shut down.")
        }
      }
    })
  } else {
    context.reply(addBlock(receivedBlockInfo))
  }

其中 addBlock方法源碼如下：

/** Add new blocks for the given stream */
private def addBlock(receivedBlockInfo: ReceivedBlockInfo): Boolean = {
  receivedBlockTracker.addBlock(receivedBlockInfo)
}

 

其中，org.apache.spark.streaming.scheduler.ReceivedBlockTracker#addBlock 源碼如下：

/** Add received block. This event will get written to the write ahead log (if enabled). */
def addBlock(receivedBlockInfo: ReceivedBlockInfo): Boolean = {
  try {
    val writeResult = writeToLog(BlockAdditionEvent(receivedBlockInfo))
    if (writeResult) {
      synchronized {
        getReceivedBlockQueue(receivedBlockInfo.streamId) += receivedBlockInfo
      }
      logDebug(s"Stream ${receivedBlockInfo.streamId} received " +
        s"block ${receivedBlockInfo.blockStoreResult.blockId}")
    } else {
      logDebug(s"Failed to acknowledge stream ${receivedBlockInfo.streamId} receiving " +
        s"block ${receivedBlockInfo.blockStoreResult.blockId} in the Write Ahead Log.")
    }
    writeResult
  } catch {
    case NonFatal(e) =>
      logError(s"Error adding block $receivedBlockInfo", e)
      false
  }
}
/** Write an update to the tracker to the write ahead log */
private def writeToLog(record: ReceivedBlockTrackerLogEvent): Boolean = {
  if (isWriteAheadLogEnabled) {
    logTrace(s"Writing record: $record")
    try {
      writeAheadLogOption.get.write(ByteBuffer.wrap(Utils.serialize(record)),
        clock.getTimeMillis())
      true
    } catch {
      case NonFatal(e) =>
        logWarning(s"Exception thrown while writing record: $record to the WriteAheadLog.", e)
        false
    }
  } else {
    true
  }
}
/** Get the queue of received blocks belonging to a particular stream */
private def getReceivedBlockQueue(streamId: Int): ReceivedBlockQueue = {
  streamIdToUnallocatedBlockQueues.getOrElseUpdate(streamId, new ReceivedBlockQueue)
}

上述代碼，主要是將BlockAdditionEvent寫WAL和更新隊列（其實就是mutable.HashMap[Int, ReceivedBlockQueue]），這個隊列中存放的是streamId ->UnallocatedBlock 的映射關系

從WAL RDD 中讀取數據

createStream 源碼如下：

/**
 * Create an input stream that pulls messages from Kafka Brokers.
 * @param ssc         StreamingContext object
 * @param kafkaParams Map of kafka configuration parameters,
 *                    see http://kafka.apache.org/08/configuration.html
 * @param topics      Map of (topic_name -> numPartitions) to consume. Each partition is consumed
 *                    in its own thread.
 * @param storageLevel Storage level to use for storing the received objects
 * @tparam K type of Kafka message key
 * @tparam V type of Kafka message value
 * @tparam U type of Kafka message key decoder
 * @tparam T type of Kafka message value decoder
 * @return DStream of (Kafka message key, Kafka message value)
 */
def createStream[K: ClassTag, V: ClassTag, U <: Decoder[_]: ClassTag, T <: Decoder[_]: ClassTag](
    ssc: StreamingContext,
    kafkaParams: Map[String, String],
    topics: Map[String, Int],
    storageLevel: StorageLevel
  ): ReceiverInputDStream[(K, V)] = {
// 可以通過設置spark.streaming.receiver.writeAheadLog.enable參數為 true來開啟WAL
  val walEnabled = WriteAheadLogUtils.enableReceiverLog(ssc.conf)
  new KafkaInputDStream[K, V, U, T](ssc, kafkaParams, topics, walEnabled, storageLevel)
}

創建的是KafkaInputDStream對象：

/**
 * Input stream that pulls messages from a Kafka Broker.
 *
 * @param kafkaParams Map of kafka configuration parameters.
 *                    See: http://kafka.apache.org/configuration.html
 * @param topics Map of (topic_name -> numPartitions) to consume. Each partition is consumed
 * in its own thread.
 * @param storageLevel RDD storage level.
 */
private[streaming]
class KafkaInputDStream[
  K: ClassTag,
  V: ClassTag,
  U <: Decoder[_]: ClassTag,
  T <: Decoder[_]: ClassTag](
    ssc_ : StreamingContext,
    kafkaParams: Map[String, String],
    topics: Map[String, Int],
    useReliableReceiver: Boolean,
    storageLevel: StorageLevel
  ) extends ReceiverInputDStream[(K, V)](ssc_) with Logging {

  def getReceiver(): Receiver[(K, V)] = {
    if (!useReliableReceiver) { // 未啟用 WAL，會使用 KafkaReceiver 對象
      new KafkaReceiver[K, V, U, T](kafkaParams, topics, storageLevel)
    } else { // 如果啟用了WAL， 使用ReliableKafkaReceiver
      new ReliableKafkaReceiver[K, V, U, T](kafkaParams, topics, storageLevel)
    }
  }
}

 

org.apache.spark.streaming.kafka.KafkaInputDStream 繼承父類的 compute方法：

/**
 * Generates RDDs with blocks received by the receiver of this stream. */
override def compute(validTime: Time): Option[RDD[T]] = {
  val blockRDD = {

    if (validTime < graph.startTime) {
      // If this is called for any time before the start time of the context,
      // then this returns an empty RDD. This may happen when recovering from a
      // driver failure without any write ahead log to recover pre-failure data.
      new BlockRDD[T](ssc.sc, Array.empty)
    } else {
      // Otherwise, ask the tracker for all the blocks that have been allocated to this stream
      // for this batch
      val receiverTracker = ssc.scheduler.receiverTracker
      val blockInfos = receiverTracker.getBlocksOfBatch(validTime).getOrElse(id, Seq.empty)

      // Register the input blocks information into InputInfoTracker
      val inputInfo = StreamInputInfo(id, blockInfos.flatMap(_.numRecords).sum)
      ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)

      // Create the BlockRDD
      createBlockRDD(validTime, blockInfos)
    }
  }
  Some(blockRDD)
}

getBlocksOfBatch 如下：

/** Get the blocks for the given batch and all input streams. */
def getBlocksOfBatch(batchTime: Time): Map[Int, Seq[ReceivedBlockInfo]] = {
  receivedBlockTracker.getBlocksOfBatch(batchTime)
}
調用：
/** Get the blocks allocated to the given batch. */
def getBlocksOfBatch(batchTime: Time): Map[Int, Seq[ReceivedBlockInfo]] = synchronized {
  timeToAllocatedBlocks.get(batchTime).map { _.streamIdToAllocatedBlocks }.getOrElse(Map.empty)
}

JobGenerator將WAL block 分配給一個batch，並生成job

取出WAL block 信息

在 org.apache.spark.streaming.scheduler.JobGenerator 中聲明了一個定時器：

// timer 會按照批次間隔 生成 GenerateJobs 任務，並放入eventLoop 堵塞隊列中
private val timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds,
  longTime => eventLoop.post(GenerateJobs(new Time(longTime))), "JobGenerator")

EventLoop 實例化代碼如下：

eventLoop = new EventLoop[JobGeneratorEvent]("JobGenerator") {
  override protected def onReceive(event: JobGeneratorEvent): Unit = processEvent(event)

  override protected def onError(e: Throwable): Unit = {
    jobScheduler.reportError("Error in job generator", e)
  }
}
eventLoop.start()

EventLoop里定義了一個LinkedBlockingDeque雙端堵塞隊列和一個執行daemon線程，daemon線程會不停從 雙端堵塞隊列中堵塞式取數據，一旦取到數據，會調 onReceive 方法，即 processEvent 方法：

/** Processes all events */
private def processEvent(event: JobGeneratorEvent) {
  logDebug("Got event " + event)
  event match {
    case GenerateJobs(time) => generateJobs(time)
    case ClearMetadata(time) => clearMetadata(time)
    case DoCheckpoint(time, clearCheckpointDataLater) =>
      doCheckpoint(time, clearCheckpointDataLater)
    case ClearCheckpointData(time) => clearCheckpointData(time)
  }
}

由於是GenerateJobs 事件， 會繼續調用generateJobs 方法：

/** Generate jobs and perform checkpoint for the given `time`.  */
private def generateJobs(time: Time) {
  // Set the SparkEnv in this thread, so that job generation code can access the environment
  // Example: BlockRDDs are created in this thread, and it needs to access BlockManager
  // Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed.
  SparkEnv.set(ssc.env)
  Try {
// 1. 將 WAL block 信息 分配給batch（這些數據塊信息是worker 節點cache 到WAL 之后發送給driver 端的）
    jobScheduler.receiverTracker.allocateBlocksToBatch(time)
// 2. 使用分配的block數據塊來生成任務
    graph.generateJobs(time) // generate jobs using allocated block
  } match {
    case Success(jobs) =>
      val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time)
      jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos))
    case Failure(e) =>
      jobScheduler.reportError("Error generating jobs for time " + time, e)
  }
// 發布DoCheckpoint 事件，保存checkpoint操作，主要是將新的checkpoint 數據寫入到 hdfs， 刪除舊的 checkpoint 數據
  eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false))
}

第一步中調用的
org.apache.spark.streaming.scheduler.ReceiverTracker#allocateBlocksToBatch方法如下：

/** Allocate all unallocated blocks to the given batch. */
def allocateBlocksToBatch(batchTime: Time): Unit = {
  if (receiverInputStreams.nonEmpty) {
    receivedBlockTracker.allocateBlocksToBatch(batchTime)
  }
}

其中，org.apache.spark.streaming.scheduler.ReceivedBlockTracker#allocateBlocksToBatch 方法如下：

def allocateBlocksToBatch(batchTime: Time): Unit = synchronized {
  if (lastAllocatedBatchTime == null || batchTime > lastAllocatedBatchTime) {
// 遍歷輸入流，根據流的 streamId 獲取未被分配的block隊列，並返回[streamId, seq[receivedBlockInfo]],由此可知，到此為止，數據其實已經從receiver中讀出來了。
   // 獲取 streamid和 WAL的blocks 的映射關系
val streamIdToBlocks = streamIds.map { streamId =>
        (streamId, getReceivedBlockQueue(streamId).dequeueAll(x => true))
    }.toMap
    val allocatedBlocks = AllocatedBlocks(streamIdToBlocks)
    if (writeToLog(BatchAllocationEvent(batchTime, allocatedBlocks))) {
      timeToAllocatedBlocks.put(batchTime, allocatedBlocks)
      lastAllocatedBatchTime = batchTime
    } else {
      logInfo(s"Possibly processed batch $batchTime need to be processed again in WAL recovery")
    }
  } else {
    // This situation occurs when:
    // 1. WAL is ended with BatchAllocationEvent, but without BatchCleanupEvent,
    // possibly processed batch job or half-processed batch job need to be processed again,
    // so the batchTime will be equal to lastAllocatedBatchTime.
    // 2. Slow checkpointing makes recovered batch time older than WAL recovered
    // lastAllocatedBatchTime.
    // This situation will only occurs in recovery time.
    logInfo(s"Possibly processed batch $batchTime need to be processed again in WAL recovery")
  }
}

其中，getReceivedBlockQueue的源碼如下：

/** Get the queue of received blocks belonging to a particular stream */
private def getReceivedBlockQueue(streamId: Int): ReceivedBlockQueue = {
  streamIdToUnallocatedBlockQueues.getOrElseUpdate(streamId, new ReceivedBlockQueue)
}

可以看到，worker node 發送過來的block 數據被取出來了。

 

根據WAL block創建 RDD

org.apache.spark.streaming.dstream.ReceiverInputDStream#createBlockRDD 源碼如下：

private[streaming] def createBlockRDD(time: Time, blockInfos: Seq[ReceivedBlockInfo]): RDD[T] = {

  if (blockInfos.nonEmpty) {
    val blockIds = blockInfos.map { _.blockId.asInstanceOf[BlockId] }.toArray
   // 所有的block已經有了WriteAheadLogRecordHandle， 創建一個WALBackedBlockRDD即可， 否則創建BlockRDD。
// 其中，WriteAheadLogRecordHandle 是一個跟WAL 相關聯的EntryInfo，實現類FileBasedWriteAheadLogSegment就包含了WAL segment 的path， offset 以及 length 信息。RDD 在真正需要數據時，根據這些handle信息從 WAL 中讀取數據。
    // Are WAL record handles present with all the blocks
    val areWALRecordHandlesPresent = blockInfos.forall { _.walRecordHandleOption.nonEmpty }

    if (areWALRecordHandlesPresent) {
      // If all the blocks have WAL record handle, then create a WALBackedBlockRDD
      val isBlockIdValid = blockInfos.map { _.isBlockIdValid() }.toArray
      val walRecordHandles = blockInfos.map { _.walRecordHandleOption.get }.toArray
      new WriteAheadLogBackedBlockRDD[T](
        ssc.sparkContext, blockIds, walRecordHandles, isBlockIdValid)
    } else {
      // Else, create a BlockRDD. However, if there are some blocks with WAL info but not
      // others then that is unexpected and log a warning accordingly.
      if (blockInfos.find(_.walRecordHandleOption.nonEmpty).nonEmpty) {
        if (WriteAheadLogUtils.enableReceiverLog(ssc.conf)) {
          logError("Some blocks do not have Write Ahead Log information; " +
            "this is unexpected and data may not be recoverable after driver failures")
        } else {
          logWarning("Some blocks have Write Ahead Log information; this is unexpected")
        }
      }
      val validBlockIds = blockIds.filter { id =>
        ssc.sparkContext.env.blockManager.master.contains(id)
      }
      if (validBlockIds.size != blockIds.size) {
        logWarning("Some blocks could not be recovered as they were not found in memory. " +
          "To prevent such data loss, enabled Write Ahead Log (see programming guide " +
          "for more details.")
      }
      new BlockRDD[T](ssc.sc, validBlockIds)
    }
  } else {
    // If no block is ready now, creating WriteAheadLogBackedBlockRDD or BlockRDD
    // according to the configuration
    if (WriteAheadLogUtils.enableReceiverLog(ssc.conf)) {
      new WriteAheadLogBackedBlockRDD[T](
        ssc.sparkContext, Array.empty, Array.empty, Array.empty)
    } else {
      new BlockRDD[T](ssc.sc, Array.empty)
    }
  }
}

org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD#compute 的源碼如下：

/**
 * Gets the partition data by getting the corresponding block from the block manager.
 * If the block does not exist, then the data is read from the corresponding record
 * in write ahead log files.
 */
override def compute(split: Partition, context: TaskContext): Iterator[T] = {
  assertValid()
  val hadoopConf = broadcastedHadoopConf.value
  val blockManager = SparkEnv.get.blockManager
  val partition = split.asInstanceOf[WriteAheadLogBackedBlockRDDPartition]
  val blockId = partition.blockId

  def getBlockFromBlockManager(): Option[Iterator[T]] = {
    blockManager.get(blockId).map(_.data.asInstanceOf[Iterator[T]])
  }

  def getBlockFromWriteAheadLog(): Iterator[T] = {
    var dataRead: ByteBuffer = null
    var writeAheadLog: WriteAheadLog = null
    try {
      // The WriteAheadLogUtils.createLog*** method needs a directory to create a
      // WriteAheadLog object as the default FileBasedWriteAheadLog needs a directory for
      // writing log data. However, the directory is not needed if data needs to be read, hence
      // a dummy path is provided to satisfy the method parameter requirements.
      // FileBasedWriteAheadLog will not create any file or directory at that path.
      // FileBasedWriteAheadLog will not create any file or directory at that path. Also,
      // this dummy directory should not already exist otherwise the WAL will try to recover
      // past events from the directory and throw errors.
      val nonExistentDirectory = new File(
        System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString).getAbsolutePath
      writeAheadLog = WriteAheadLogUtils.createLogForReceiver(
        SparkEnv.get.conf, nonExistentDirectory, hadoopConf)
      dataRead = writeAheadLog.read(partition.walRecordHandle)
    } catch {
      case NonFatal(e) =>
        throw new SparkException(
          s"Could not read data from write ahead log record ${partition.walRecordHandle}", e)
    } finally {
      if (writeAheadLog != null) {
        writeAheadLog.close()
        writeAheadLog = null
      }
    }
    if (dataRead == null) {
      throw new SparkException(
        s"Could not read data from write ahead log record ${partition.walRecordHandle}, " +
          s"read returned null")
    }
    logInfo(s"Read partition data of $this from write ahead log, record handle " +
      partition.walRecordHandle)
    if (storeInBlockManager) {
      blockManager.putBytes(blockId, dataRead, storageLevel)
      logDebug(s"Stored partition data of $this into block manager with level $storageLevel")
      dataRead.rewind()
    }
    blockManager.dataDeserialize(blockId, dataRead).asInstanceOf[Iterator[T]]
  }
 // 如果partition.isBlockIdValid 為true，則說明該 block 數據存在executors 中
  if (partition.isBlockIdValid) {
// 先根據 BlockManager從 executor中讀取數據， 如果沒有，再從WAL 中讀取數據
// BlockManager 從內存還是從磁盤上獲取的數據 ？
blockManager 從 local 或 remote 獲取 block，其中 local既可以從 memory 中獲取也可以從 磁盤中讀取， 其中remote獲取數據是同步的，即在fetch block 過程中會一直blocking。
    getBlockFromBlockManager().getOrElse { getBlockFromWriteAheadLog() }
  } else {
    getBlockFromWriteAheadLog()
  }
}

 

至此，從啟動 receiver，到receiver 接收數據並保存到WAL block，driver 接收WAL 的block 信息，直到spark streaming 通過WAL RDD 來獲取數據等等都一一做了說明。