一、生产者客户端配置参数acks说明
首先，客户端需要配置一个
acks参数，默认值是1，下面是acks各个值的说明
acks=-1，太慢，acks=0，有风险，acks=1，则是推荐，所以也是默认值的原因
1、acks=1
这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。如果follower没有成功备份数据，而此时leader又无法提供服务，则消息会丢失。
2、acks=0
表示producer不需要等待任何确认收到的信息，副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为-1。
3、acks=-1
这意味着leader需要等待ISR中所有备份都成功写入日志。只要任何一个备份存活，数据都不会丢失。min.insync.replicas指定必须确认写入才能被认为成功的副本的最小数量。
二、请求在写入Leader的数据管道之前，则会验证Leader的ISR副本数量和配置中的最小ISR数量
def appendRecordsToLeader(records: MemoryRecords, origin: AppendOrigin, requiredAcks: Int,
requestLocal: RequestLocal): LogAppendInfo = {
//函数首先获取leaderIsrUpdateLock的读锁,以确保对Leader和ISR（In-Sync Replica）的更新操作是同步的。
val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
//然后检查当前是否有Leader日志，
leaderLogIfLocal match {
//如果存在Leader日志，
case Some(leaderLog) =>
//则获取最小ISR（MinInSyncReplicas）的配置和ISR的大小。
val minIsr = leaderLog.config.minInSyncReplicas
val inSyncSize = partitionState.isr.size
// Avoid writing to leader if there are not enough insync replicas to make it safe，如果没有足够的不同步副本来使其安全，请避免写入领导者
//如果ISR的大小小于最小ISR要求，并且requiredAcks的值为-1（表示不需要确认），则抛出NotEnoughReplicasException异常。
if (inSyncSize < minIsr && requiredAcks == -1) {
throw new NotEnoughReplicasException(s"The size of the current ISR ${partitionState.isr} " +
s"is insufficient to satisfy the min.isr requirement of $minIsr for partition $topicPartition")
}
//调用Leader日志的appendAsLeader方法将记录作为Leader追加到日志中，并传递相关参数。
val info = leaderLog.appendAsLeader(records, leaderEpoch = this.leaderEpoch, origin,
interBrokerProtocolVersion, requestLocal)
// we may need to increment high watermark since ISR could be down to 1,
// 我们可能需要增加高水位线，因为 ISR 可能降至 1
(info, maybeIncrementLeaderHW(leaderLog))
//如果没有，则抛出NotLeaderOrFollowerException异常。
case None =>
throw new NotLeaderOrFollowerException("Leader not local for partition %s on broker %d"
.format(topicPartition, localBrokerId))
}
}
//返回追加记录的信息，并根据是否增加了Leader高水位线，将LeaderHwChange.INCREASED或LeaderHwChange.SAME复制给返回信息的副本。
info.copy(if (leaderHWIncremented) LeaderHwChange.INCREASED else LeaderHwChange.SAME)
}
1、Leader的ISR小于配置文件中minInSyncReplicas，并且acks=-1，则抛异常
会验证acks=-1并且当前Leader的ISR副本数量小于配置中规定的最小值
val minIsr = leaderLog.config.minInSyncReplicas
val inSyncSize = partitionState.isr.size
if (inSyncSize < minIsr && requiredAcks == -1) {
throw new NotEnoughReplicasException(s"The size of the current ISR ${partitionState.isr} " +
s"is insufficient to satisfy the min.isr requirement of $minIsr for partition $topicPartition")
}
2、如果acks不等于-1，则就算Leader的ISR小于配置，也会正常执行写入数据管道操作
//调用Leader日志的appendAsLeader方法将记录作为Leader追加到日志中，并传递相关参数。
val info = leaderLog.appendAsLeader(records, leaderEpoch = this.leaderEpoch, origin,
interBrokerProtocolVersion, requestLocal)
三、请求把数据写入到Leader的数据管道后，acks=-1和非-1，有不同的逻辑
这里不从头开始，如果想知道推送的数据怎么到下面方法的，可以看kafka 3.5 kafka服务端接收生产者发送的数据源码
/**
*将消息附加到分区的领导副本，并等待它们复制到其他副本;当超时或满足所需的 ACK 时，将触发回调函数;如果回调函数本身已经在某个对象上同步，则传递此对象以避免死锁
* 请注意，所有挂起的延迟检查操作都存储在队列中。所有 ReplicaManager.appendRecords（） 的调用方都应为所有受影响的分区调用 ActionQueue.tryCompleteActions，而不会保留任何冲突的锁。
*/
def appendRecords(timeout: Long,
requiredAcks: Short,
internalTopicsAllowed: Boolean,
origin: AppendOrigin,
entriesPerPartition: Map[TopicPartition, MemoryRecords],
responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
delayedProduceLock: Option[Lock] = None,
recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching): Unit = {
//省略代码
//把数据中加入到本地Log
val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
origin, entriesPerPartition, requiredAcks, requestLocal)
//省略代码
//调用recordConversionStatsCallback方法，将每个分区的记录转换统计信息传递给回调函数。
recordConversionStatsCallback(localProduceResults.map { case (k, v) => k -> v.info.recordConversionStats })
//通过 delayedProduceRequestRequired 方法判断是否需要等待其它副本完成写入，如果 acks = -1，则需要等待所有副本的回应
if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
//根据条件判断是否需要创建延迟的produce操作。如果需要，创建一个DelayedProduce对象，并将它添加到delayedProducePurgatory中。
val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)
//创建（主题、分区）对的列表，以用作此延迟生成操作的键
val producerRequestKeys = entriesPerPartition.keys.map(TopicPartitionOperationKey(_)).toSeq
// 再一次尝试完成该延时请求
// 如果暂时无法完成，则将对象放入到相应的Purgatory中等待后续处理
delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)
} else {
//如果不需要延迟操作，直接将produce的结果返回给回调函数。
val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus }
responseCallback(produceResponseStatus)
}else{
//如果不需要延迟操作，直接将produce的结果返回给回调函数。
// we can respond immediately
val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus }
responseCallback(produceResponseStatus)
}
//省略代码
}
1、如果acks=-1，则会创建延迟Produce请求，等待ISR中所有副本的响应
//它用于判断是否需要延迟发送生产请求并等待复制完成
// 1. required acks = -1 判断requiredAcks是否等于-1，即是否需要等待所有副本的确认。
// 2. there is data to append 判断entriesPerPartition是否不为空，即是否有要追加的数据。
// 3. at least one partition append was successful (fewer errors than partitions) 计算localProduceResults中异常定义的数量，判断其是否小于entriesPerPartition的大小，即是否至少有一个分区的追加操作成功（即比分区数少的错误，如果全错，就应该直接返回）。
private def delayedProduceRequestRequired(requiredAcks: Short,
entriesPerPartition: Map[TopicPartition, MemoryRecords],
localProduceResults: Map[TopicPartition, LogAppendResult]): Boolean = {
requiredAcks == -1 &&
entriesPerPartition.nonEmpty &&
localProduceResults.values.count(_.exception.isDefined) < entriesPerPartition.size
}
/**
*检查操作是否可以完成，如果没有，则根据给定的监视键进行监视
*请注意，可以在多个密钥上监视延迟操作。对于某些键（但不是所有键），操作可能会在添加到监视列表后完成。
* 在这种情况下，操作被视为已完成，不会添加到其余键的监视列表中。过期收割线程将从存在该操作的任何观察程序列表中删除此操作。
* @param operation the delayed operation to be checked 要检查的延迟操作
* @param watchKeys keys for bookkeeping the operation 用于监视的键
* @return true iff the delayed operations can be completed by the caller 如果延迟操作可以由调用方完成，则为 true
*/
def tryCompleteElseWatch(operation: T, watchKeys: Seq[Any]): Boolean = {
assert(watchKeys.nonEmpty, "The watch key list can't be empty")
//尝试完成操作,如果操作不能立即完成，则将操作添加到所有观察键的观察列表中，并递增estimatedTotalOperations计数器的值
if (operation.safeTryCompleteOrElse {
watchKeys.foreach(key => watchForOperation(key, operation))
if (watchKeys.nonEmpty) estimatedTotalOperations.incrementAndGet()
}) return true
//如果操作仍未完成，则根据条件执行以下操作：
if (!operation.isCompleted) {
//如果启用了定时器（timerEnabled为真），则将操作添加到超时定时器中。
if (timerEnabled)
timeoutTimer.add(operation)
//如果操作已完成，则取消定时器任务。
if (operation.isCompleted) {
// cancel the timer task
operation.cancel()
}
}
//返回false表示操作未完成。
false
}
2、如果acks不等于-1，写入到Leader的数据管道后，则直接执行回调函数返回结果
//如果不需要延迟操作，直接将produce的结果返回给回调函数。
// we can respond immediately
val produceResponseStatus = produceStatus.map { case (k, status) => k -> status.responseStatus }
responseCallback(produceResponseStatus)
四、在返回response时，回调函数会遍历分区异常信息
//用于发送 produce 响应的回调 ProduceResponse 的构造能够接受自动生成的协议数据，因此 KafkaApishandleProduceRequest 应应用自动生成的协议以避免额外的转换
@nowarn("cat=deprecation")
def sendResponseCallback(responseStatus: Map[TopicPartition, PartitionResponse]): Unit = {
val mergedResponseStatus = responseStatus ++ unauthorizedTopicResponses ++ nonExistingTopicResponses ++ invalidRequestResponses
var errorInResponse = false
mergedResponseStatus.forKeyValue { (topicPartition, status) =>
if (status.error != Errors.NONE) {
errorInResponse = true
debug("Produce request with correlation id %d from client %s on partition %s failed due to %s".format(
request.header.correlationId,
request.header.clientId,
topicPartition,
status.error.exceptionName))
}
}
//记录带宽和请求配额特定的值，并在违反任何配额时通过静音通道来限制。如果违反了两个配额，请使用两个配额之间的最大限制时间。请注意，如果 acks == 0，则不会强制执行请求配额。
val timeMs = time.milliseconds()
val requestSize = request.sizeInBytes
val bandwidthThrottleTimeMs = quotas.produce.maybeRecordAndGetThrottleTimeMs(request, requestSize, timeMs)
val requestThrottleTimeMs =
if (produceRequest.acks == 0) 0
else quotas.request.maybeRecordAndGetThrottleTimeMs(request, timeMs)
val maxThrottleTimeMs = Math.max(bandwidthThrottleTimeMs, requestThrottleTimeMs)
if (maxThrottleTimeMs > 0) {
request.apiThrottleTimeMs = maxThrottleTimeMs
if (bandwidthThrottleTimeMs > requestThrottleTimeMs) {
requestHelper.throttle(quotas.produce, request, bandwidthThrottleTimeMs)
} else {
requestHelper.throttle(quotas.request, request, requestThrottleTimeMs)
}
}
//如果produceRequest.acks等于0，表示不需要响应。
if (produceRequest.acks == 0) {
//如果生产者请求，则无需操作;
//但是，如果在处理请求时出现任何错误，由于生产者期望没有响应，服务器将关闭套接字服务器，以便生产者客户端知道发生了一些错误并刷新其元数据
if (errorInResponse) {
//如果errorInResponse为true，则关闭连接并发送包含错误信息的ProduceResponse响应
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"
)
requestChannel.closeConnection(request, new ProduceResponse(mergedResponseStatus.asJava).errorCounts)
} else {
//如果没有异常，发送无操作的响应。
requestHelper.sendNoOpResponseExemptThrottle(request)
}
} else {
//如果produceRequest.acks不等于0，将mergedResponseStatus和maxThrottleTimeMs作为参数构造ProduceResponse响应，并通过requestChannel发送响应。
requestChannel.sendResponse(request, new ProduceResponse(mergedResponseStatus.asJava, maxThrottleTimeMs), None)
}
}
1、如果acks=0，则关闭套接字服务器
//如果生产者请求，则无需操作;
//但是，如果在处理请求时出现任何错误，由于生产者期望没有响应，服务器将关闭套接字服务器，以便生产者客户端知道发生了一些错误并刷新其元数据
if (errorInResponse) {
//如果errorInResponse为true，则关闭连接并发送包含错误信息的ProduceResponse响应
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"
)
requestChannel.closeConnection(request, new ProduceResponse(mergedResponseStatus.asJava).errorCounts)
}
2、如果acks不等0，会返回异常信息
requestChannel.sendResponse(request, new ProduceResponse(mergedResponseStatus.asJava, maxThrottleTimeMs), None)
