kafka学习六-生产延迟操作

caseApiKeys.PRODUCE=>handleProduceRequest(request)

/*** Handle a produce request* 处理生产请求*/defhandleProduceRequest(request: RequestChannel.Request) {
//拿到生产者请求的内容valproduceRequest=request.body[ProduceRequest]
//追加大小valnumBytesAppended=request.header.toStruct.sizeOf+request.sizeOfBodyInBytes// the callback for sending a produce response//发送生产响应的回调defsendResponseCallback(responseStatus: Map[TopicPartition, PartitionResponse]) {
valmergedResponseStatus=responseStatus++unauthorizedTopicResponses++nonExistingTopicResponsesvarerrorInResponse=false//生产响应回调defproduceResponseCallback(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 metadataif (errorInResponse) {
valexceptionsSummary=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, newProduceResponse(mergedResponseStatus.asJava).errorCounts)
        } else {
//发送没有操作响应的操作//发送NoOp响应豁免节流阀sendNoOpResponseExemptThrottle(request)
        }
      } else { //ack =1 //发送响应sendResponseMaybeThrottle(request, requestThrottleMs=>newProduceResponse(mergedResponseStatus.asJava, bandwidthThrottleTimeMs+requestThrottleMs))
      }
    }
//acks =-1的情况// When this callback is triggered, the remote API call has completedrequest.apiRemoteCompleteTimeNanos=time.nanosecondsquotas.produce.maybeRecordAndThrottle(
request.session.sanitizedUser,
request.header.clientId,
numBytesAppended,
produceResponseCallback)
  }
//处理统计回调defprocessingStatsCallback(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更改了一些被限制的指标（产生/消耗的字节，请求处理时间等）如果违反配额，//则在延迟后调用回调，否则立即调用回调。节流时间计算可能被覆盖 子类。defmaybeRecordAndThrottle(sanitizedUser: String, clientId: String, value: Double, callback: Int=>Unit): Int= {
if (quotasEnabled) {
//获取或创建QuotaSensorsvalclientSensors=getOrCreateQuotaSensors(sanitizedUser, clientId)
recordAndThrottleOnQuotaViolation(clientSensors, value, callback)
  } else {
// Don't record any metrics if quotas are not enabled at any levelvalthrottleTimeMs=0callback(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满足时，将触发回调函数；* 如果回调函数本身已经在某个对象上同步，则传递此对象以避免死锁。*/defappendRecords(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)) {
valsTime=time.milliseconds//追加到本地logvallocalProduceResults=appendToLocalLog(internalTopicsAllowed=internalTopicsAllowed,
isFromClient=isFromClient, entriesPerPartition, requiredAcks)
debug("Produce to local log in %d ms".format(time.milliseconds-sTime))
valproduceStatus=localProduceResults.map { case (topicPartition, result) =>topicPartition->ProducePartitionStatus(
result.info.lastOffset+1, // required offsetnewPartitionResponse(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//创建延迟生产操作valproduceMetadata=ProduceMetadata(requiredAcks, produceStatus)
//创建新的延迟生产valdelayedProduce=newDelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)
// create a list of (topic, partition) pairs to use as keys for this delayed produce operation//创建（主题，分区）对的列表，以用作此延迟的生产操作的键valproducerRequestKeys=entriesPerPartition.keys.map(newTopicPartitionOperationKey(_)).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 immediatelyvalproduceResponseStatus=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 allvalresponseStatus=entriesPerPartition.map { case (topicPartition, _) =>topicPartition->newPartitionResponse(Errors.INVALID_REQUIRED_ACKS,
LogAppendInfo.UnknownLogAppendInfo.firstOffset, RecordBatch.NO_TIMESTAMP, LogAppendInfo.UnknownLogAppendInfo.logStartOffset)
     }
//响应回调responseCallback(responseStatus)
   }
 }

classDelayedProduce(delayMs: Long,
produceMetadata: ProduceMetadata,
replicaManager: ReplicaManager,
responseCallback: Map[TopicPartition, PartitionResponse] =>Unit,
lockOpt: Option[Lock] =None)
extendsDelayedOperation(delayMs, lockOpt)