kafka学习六-生产延迟操作

case ApiKeys.PRODUCE => handleProduceRequest(request)

/**
 * Handle a produce request
  * 处理生产请求
 */
def handleProduceRequest(request: RequestChannel.Request) {
  //拿到生产者请求的内容
  val produceRequest = request.body[ProduceRequest]
  //追加大小
  val numBytesAppended = request.header.toStruct.sizeOf + request.sizeOfBodyInBytes

  // the callback for sending a produce response
  //发送生产响应的回调
  def sendResponseCallback(responseStatus: Map[TopicPartition, PartitionResponse]) {

    val mergedResponseStatus = responseStatus ++ unauthorizedTopicResponses ++ nonExistingTopicResponses
    var errorInResponse = false

    //生产响应回调
    def produceResponseCallback(bandwidthThrottleTimeMs: Int) {
      //生产请求的ack = 0的请求
      /**
        * 如果生产者request.required.acks = 0，则不需要任何操作； 但是，如果在处理请求时出现任何错误，
        * 由于生产者不希望响应，则服务器将关闭套接字服务器，以便生产者客户端将知道发生了一些错误并刷新其元数据
        */
      if (produceRequest.acks == 0) { //acks = 0 即不需要acks，没啥需要特别做的
        // no operation needed if producer request.required.acks = 0; however, if there is any error in handling
        // the request, since no response is expected by the producer, the server will close socket server so that
        // the producer client will know that some error has happened and will refresh its metadata
        if (errorInResponse) {
          val exceptionsSummary = mergedResponseStatus.map { case (topicPartition, status) =>
            topicPartition -> status.error.exceptionName
          }.mkString(", ")
          info(
            s"Closing connection due to error during produce request with correlation id ${request.header.correlationId} " +
              s"from client id ${request.header.clientId} with ack=0\n" +
              s"Topic and partition to exceptions: $exceptionsSummary"
          )
          //关闭连接
          closeConnection(request, new ProduceResponse(mergedResponseStatus.asJava).errorCounts)
        } else {
          //发送没有操作响应的操作
          //发送NoOp响应豁免节流阀
          sendNoOpResponseExemptThrottle(request)
        }
      } else { //ack =1 //发送响应
        sendResponseMaybeThrottle(request, requestThrottleMs =>
          new ProduceResponse(mergedResponseStatus.asJava, bandwidthThrottleTimeMs + requestThrottleMs))
      }
    }
    //acks =-1的情况
    // When this callback is triggered, the remote API call has completed
    request.apiRemoteCompleteTimeNanos = time.nanoseconds

    quotas.produce.maybeRecordAndThrottle(
      request.session.sanitizedUser,
      request.header.clientId,
      numBytesAppended,
      produceResponseCallback)
  }

  //处理统计回调
  def processingStatsCallback(processingStats: Map[TopicPartition, RecordsProcessingStats]): Unit = {
    processingStats.foreach { case (tp, info) =>
      updateRecordsProcessingStats(request, tp, info)
    }
  }

    // call the replica manager to append messages to the replicas
    //副本管理进行追加消息调用  重要
    replicaManager.appendRecords(
      timeout = produceRequest.timeout.toLong,
      requiredAcks = produceRequest.acks,
      internalTopicsAllowed = internalTopicsAllowed,
      isFromClient = true,
      entriesPerPartition = authorizedRequestInfo,
      responseCallback = sendResponseCallback,
      processingStatsCallback = processingStatsCallback)

    // if the request is put into the purgatory, it will have a held reference and hence cannot be garbage collected;
    // hence we clear its data here in order to let GC reclaim its memory since it is already appended to log
    /**
      * 如果将请求放入炼狱延迟组件，它将具有保留的引用，因此无法进行垃圾收集；
      * 因此，我们在这里清除其数据是为了让GC回收其内存，因为它已被附加到日志中
      */
    produceRequest.clearPartitionRecords()
  }
}

//记录用户/客户ID更改了一些被限制的指标（产生/消耗的字节，请求处理时间等）如果违反配额，
//则在延迟后调用回调，否则立即调用回调。节流时间计算可能被覆盖 子类。
def maybeRecordAndThrottle(sanitizedUser: String, clientId: String, value: Double, callback: Int => Unit): Int = {
  if (quotasEnabled) {
      //获取或创建QuotaSensors
    val clientSensors = getOrCreateQuotaSensors(sanitizedUser, clientId)
    recordAndThrottleOnQuotaViolation(clientSensors, value, callback)
  } else {
    // Don't record any metrics if quotas are not enabled at any level
    val throttleTimeMs = 0
    callback(throttleTimeMs)
    throttleTimeMs
  }
}

 /**
  * Append messages to leader replicas of the partition, and wait for them to be replicated to other replicas;
  * the callback function will be triggered either when timeout or the required acks are satisfied;
  * if the callback function itself is already synchronized on some object then pass this object to avoid deadlock.
   * 将消息追加到分区的leader副本，然后等待它们被复制到其他副本； 当超时或所需的acks满足时，将触发回调函数；
   * 如果回调函数本身已经在某个对象上同步，则传递此对象以避免死锁。
  */
 def appendRecords(timeout: Long,
                   requiredAcks: Short,
                   internalTopicsAllowed: Boolean,
                   isFromClient: Boolean,
                   entriesPerPartition: Map[TopicPartition, MemoryRecords],
                   responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
                   delayedProduceLock: Option[Lock] = None,
                   processingStatsCallback: Map[TopicPartition, RecordsProcessingStats] => Unit = _ => ()) {
   if (isValidRequiredAcks(requiredAcks)) {
     val sTime = time.milliseconds
     //追加到本地log
     val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
       isFromClient = isFromClient, entriesPerPartition, requiredAcks)
     debug("Produce to local log in %d ms".format(time.milliseconds - sTime))

     val produceStatus = localProduceResults.map { case (topicPartition, result) =>
       topicPartition ->
               ProducePartitionStatus(
                 result.info.lastOffset + 1, // required offset
                 new PartitionResponse(result.error, result.info.firstOffset, result.info.logAppendTime, result.info.logStartOffset)) // respon
     }

     //处理统计回调函数
     processingStatsCallback(localProduceResults.mapValues(_.info.recordsProcessingStats))

     //延迟生产请求要求
     if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
       // create delayed produce operation
       //创建延迟生产操作
       val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
       //创建新的延迟生产
       val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)

       // create a list of (topic, partition) pairs to use as keys for this delayed produce operation
       //创建（主题，分区）对的列表，以用作此延迟的生产操作的键
       val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toSeq

       // try to complete the request immediately, otherwise put it into the purgatory
       // this is because while the delayed produce operation is being created, new
       // requests may arrive and hence make this operation completable.
       //尝试立即完成请求，否则将其放入炼狱，这是因为在创建延迟的生产操作时，新的请求可能会到达并因此使该操作可完成。
       delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)

     } else {
       // we can respond immediately
       val produceResponseStatus = produceStatus.mapValues(status => status.responseStatus)
       responseCallback(produceResponseStatus)
     }
   } else {
     // If required.acks is outside accepted range, something is wrong with the client
     // Just return an error and don't handle the request at all
     val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
       topicPartition -> new PartitionResponse(Errors.INVALID_REQUIRED_ACKS,
         LogAppendInfo.UnknownLogAppendInfo.firstOffset, RecordBatch.NO_TIMESTAMP, LogAppendInfo.UnknownLogAppendInfo.logStartOffset)
     }
     //响应回调
     responseCallback(responseStatus)
   }
 }  

class DelayedProduce(delayMs: Long,
                     produceMetadata: ProduceMetadata,
                     replicaManager: ReplicaManager,
                     responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
                     lockOpt: Option[Lock] = None)
  extends DelayedOperation(delayMs, lockOpt)