Kafka Broker核心模块：请求处理与副本管理深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

/**
 * BrokerServer的启动方法 - 按特定顺序初始化各个组件
 * 确保组件间的依赖关系正确建立
 */
override def startup(): Unit = {
  if (!maybeChangeStatus(SHUTDOWN, STARTING)) return
  val startupDeadline = Deadline.fromDelay(time, config.serverMaxStartupTimeMs, TimeUnit.MILLISECONDS)
  
  try {
    // 1. 启动共享服务器组件
    sharedServer.startForBroker()
    info("启动Broker")

    // 2. 初始化配置和调度器
    config.dynamicConfig.initialize()
    quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-", ProcessRole.BrokerRole.toString)
    
    kafkaScheduler = new KafkaScheduler(config.backgroundThreads)
    kafkaScheduler.startup()

    // 3. 初始化监控和元数据缓存
    brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
    logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size)
    metadataCache = new KRaftMetadataCache(config.nodeId, () => raftManager.client.kraftVersion())

    // 4. 创建日志管理器（但不启动，等待元数据追赶）
    logManager = LogManager(config,
      sharedServer.metaPropsEnsemble.errorLogDirs().asScala.toSeq,
      metadataCache,
      kafkaScheduler,
      time,
      brokerTopicStats,
      logDirFailureChannel)

    // 5. 创建生命周期管理器
    lifecycleManager = new BrokerLifecycleManager(config,
      time,
      s"broker-${config.nodeId}-",
      logDirs = logManager.directoryIdsSet,
      () => new Thread(() => shutdown(), "kafka-shutdown-thread").start())

    // 6. 创建副本管理器
    this._replicaManager = new ReplicaManager(
      config = config,
      metrics = metrics,
      time = time,
      scheduler = kafkaScheduler,
      logManager = logManager,
      remoteLogManager = remoteLogManagerOpt,
      quotaManagers = quotaManagers,
      metadataCache = metadataCache,
      logDirFailureChannel = logDirFailureChannel,
      // ... 其他参数
    )

    // 7. 创建网络服务器和请求处理组件
    socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)
    dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.nodeId, socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time, config.numIoThreads, s"data-plane-kafka-request-handler", ListenerName.normalised("data-plane"))

    // 8. 等待元数据追赶完成
    info("等待元数据追赶完成")
    FutureUtils.waitWithLogging(logger.underlying, logIdent,
      "元数据追赶完成", lifecycleManager.metadataListener.highWatermark(), startupDeadline, time)

    // 9. 启动核心组件
    logManager.startup()
    replicaManager.startup()
    
    // 10. 设置为可用状态
    FutureUtils.waitWithLogging(logger.underlying, logIdent,
      "broker准备解除隔离", lifecycleManager.setReadyToUnfence(), startupDeadline, time)

    // 11. 启用网络请求处理
    socketServer.enableRequestProcessing(authorizerFutures)

    maybeChangeStatus(STARTING, STARTED)
  } catch {
    case e: Throwable =>
      fatal("Broker启动过程中发生致命错误，准备关闭", e)
      shutdown()
      throw if (e.isInstanceOf[ExecutionException]) e.getCause else e
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

/**
 * KafkaApis - Kafka API请求的核心处理器
 * 根据请求类型分发到相应的处理方法
 */
class KafkaApis(
  val requestChannel: RequestChannel,
  val replicaManager: ReplicaManager,
  val groupCoordinator: GroupCoordinator,
  val txnCoordinator: TransactionCoordinator,
  // ... 其他依赖组件
) extends ApiRequestHandler with Logging {

  /**
   * 请求分发的核心方法
   * 根据API Key路由到对应的处理方法
   */
  def handle(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
    try {
      trace(s"处理 ${request.header.apiKey} 请求从 ${request.context.clientAddress}")
      
      // 根据API类型分发请求
      request.header.apiKey match {
        case ApiKeys.PRODUCE => 
          handleProduceRequest(request, requestLocal)
        case ApiKeys.FETCH => 
          handleFetchRequest(request)
        case ApiKeys.LIST_OFFSETS => 
          handleListOffsetRequest(request)
        case ApiKeys.METADATA => 
          handleTopicMetadataRequest(request)
        case ApiKeys.OFFSET_COMMIT => 
          handleOffsetCommitRequest(request, requestLocal).exceptionally(handleError)
        case ApiKeys.OFFSET_FETCH => 
          handleOffsetFetchRequest(request).exceptionally(handleError)
        case ApiKeys.FIND_COORDINATOR => 
          handleFindCoordinatorRequest(request)
        case ApiKeys.JOIN_GROUP => 
          handleJoinGroupRequest(request, requestLocal).exceptionally(handleError)
        case ApiKeys.HEARTBEAT => 
          handleHeartbeatRequest(request).exceptionally(handleError)
        case ApiKeys.LEAVE_GROUP => 
          handleLeaveGroupRequest(request).exceptionally(handleError)
        case ApiKeys.SYNC_GROUP => 
          handleSyncGroupRequest(request, requestLocal).exceptionally(handleError)
        case ApiKeys.CREATE_TOPICS => 
          forwardToController(request) // 转发给Controller
        case ApiKeys.DELETE_TOPICS => 
          forwardToController(request)
        case ApiKeys.INIT_PRODUCER_ID => 
          handleInitProducerIdRequest(request, requestLocal)
        case ApiKeys.ADD_PARTITIONS_TO_TXN => 
          handleAddPartitionsToTxnRequest(request, requestLocal)
        case ApiKeys.END_TXN => 
          handleEndTxnRequest(request, requestLocal)
        case _ => 
          throw new IllegalStateException(s"未知的API Key: ${request.header.apiKey}")
      }
    } catch {
      case e: FatalExitError => 
        throw e
      case e: Throwable => 
        error(s"处理请求时发生未捕获的异常", e)
        requestHelper.handleError(request, e)
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

/**
 * 处理Producer发送的Produce请求
 * 包括权限验证、记录验证和写入副本等步骤
 */
def handleProduceRequest(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
  val produceRequest = request.body[ProduceRequest]

  // 1. 事务权限验证
  if (RequestUtils.hasTransactionalRecords(produceRequest)) {
    val isAuthorizedTransactional = produceRequest.transactionalId != null &&
      authHelper.authorize(request.context, WRITE, TRANSACTIONAL_ID, produceRequest.transactionalId)
    if (!isAuthorizedTransactional) {
      requestHelper.sendErrorResponseMaybeThrottle(request, Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED.exception)
      return
    }
  }

  // 2. 收集和验证分区数据
  val unauthorizedTopicResponses = mutable.Map[TopicIdPartition, PartitionResponse]()
  val nonExistingTopicResponses = mutable.Map[TopicIdPartition, PartitionResponse]()
  val invalidRequestResponses = mutable.Map[TopicIdPartition, PartitionResponse]()
  val authorizedRequestInfo = mutable.Map[TopicIdPartition, MemoryRecords]()

  // 解析请求数据
  produceRequest.data.topicData.forEach { topic =>
    topic.partitionData.forEach { partition =>
      val (topicName, topicId) = if (topic.topicId().equals(Uuid.ZERO_UUID)) {
        (topic.name(), metadataCache.getTopicId(topic.name()))
      } else {
        (metadataCache.getTopicName(topic.topicId).orElse(topic.name), topic.topicId())
      }

      val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topicName, partition.index()))
      val memoryRecords = partition.records.asInstanceOf[MemoryRecords]

      // 权限检查
      if (!authorizedTopics.contains(topicName)) {
        unauthorizedTopicResponses += topicIdPartition -> new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)
      } else if (!metadataCache.contains(topicIdPartition.topicPartition)) {
        nonExistingTopicResponses += topicIdPartition -> new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)
      } else {
        try {
          // 验证记录格式
          ProduceRequest.validateRecords(request.header.apiVersion, memoryRecords)
          authorizedRequestInfo += (topicIdPartition -> memoryRecords)
        } catch {
          case e: ApiException =>
            invalidRequestResponses += topicIdPartition -> new PartitionResponse(Errors.forException(e))
        }
      }
    }
  }

  // 3. 定义响应回调函数
  def sendResponseCallback(responseStatus: Map[TopicIdPartition, PartitionResponse]): Unit = {
    val mergedResponseStatus = unauthorizedTopicResponses ++ nonExistingTopicResponses ++ 
                              invalidRequestResponses ++ responseStatus
    var errorInResponse = false

    mergedResponseStatus.forKeyValue { (topicIdPartition, status) =>
      if (status.error != Errors.NONE) {
        errorInResponse = true
        debug(s"生产请求到主题 ${topicIdPartition.topic} 分区 ${topicIdPartition.partition} 失败，错误: ${status.error}")
      }
    }

    // 发送响应
    requestChannel.sendResponse(request, 
      new ProduceResponse(mergedResponseStatus.asJava, maxThrottleTimeMs, nodeEndpoints.values.toList.asJava), None)
  }

  // 4. 委托给ReplicaManager处理实际的消息写入
  if (authorizedRequestInfo.isEmpty) {
    sendResponseCallback(Map.empty)
  } else {
    replicaManager.handleProduceAppend(
      timeout = produceRequest.timeout.toLong,
      requiredAcks = produceRequest.acks,
      internalTopicsAllowed = request.header.clientId == "__admin_client",
      transactionalId = produceRequest.transactionalId,
      entriesPerPartition = authorizedRequestInfo,
      responseCallback = sendResponseCallback,
      recordValidationStatsCallback = processingStatsCallback,
      requestLocal = requestLocal,
      transactionSupportedOperation = transactionSupportedOperation
    )

    // 清理请求数据以释放内存
    produceRequest.clearPartitionRecords()
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227

/**
 * ReplicaManager - 副本管理器的核心实现
 * 负责管理分区副本、处理读写请求和维护副本一致性
 */
class ReplicaManager(
  val config: KafkaConfig,
  metrics: Metrics,
  time: Time,
  scheduler: Scheduler,
  val logManager: LogManager,
  // ... 其他参数
) extends Logging with KafkaMetricsGroup {

  /**
   * 处理Producer的append请求
   * 将消息写入Leader副本并等待ISR副本确认
   */
  def handleProduceAppend(
    timeout: Long,
    requiredAcks: Short,
    internalTopicsAllowed: Boolean,
    transactionalId: String,
    entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
    responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit,
    recordValidationStatsCallback: Map[TopicIdPartition, RecordValidationStats] => Unit = _ => (),
    requestLocal: RequestLocal = RequestLocal.noCaching,
    transactionSupportedOperation: TransactionSupportedOperation
  ): Unit = {

    // 1. 处理事务验证（如果需要）
    val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
    val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
    
    entriesPerPartition.foreachEntry { (topicIdPartition, records) =>
      val transactionalBatches = records.batches.asScala.filter(batch => 
        batch.hasProducerId && batch.isTransactional)
      transactionalBatches.foreach(batch => 
        transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
      if (transactionalBatches.nonEmpty) {
        topicPartitionBatchInfo.put(topicIdPartition.topicPartition(), records.firstBatch.baseSequence)
      }
    }

    // 2. 定义验证后的回调处理
    def postVerificationCallback(
      newRequestLocal: RequestLocal,
      results: (Map[TopicPartition, Errors], Map[TopicPartition, VerificationGuard])
    ): Unit = {
      val (preAppendErrors, verificationGuards) = results
      
      // 构建最终的entry映射，排除有错误的分区
      val filteredEntriesPerPartition = entriesPerPartition.filter { case (topicIdPartition, _) =>
        !preAppendErrors.contains(topicIdPartition.topicPartition()) ||
        preAppendErrors(topicIdPartition.topicPartition()) == Errors.NONE
      }

      // 如果有有效的分区数据，继续处理
      if (filteredEntriesPerPartition.nonEmpty) {
        appendRecords(
          timeout = timeout,
          requiredAcks = requiredAcks,
          internalTopicsAllowed = internalTopicsAllowed,
          origin = AppendOrigin.Client,
          entriesPerPartition = filteredEntriesPerPartition,
          responseCallback = (results: Map[TopicIdPartition, PartitionResponse]) => {
            // 合并预验证错误和append结果
            val mergedResults = preAppendErrors.map { case (tp, error) =>
              new TopicIdPartition(entriesPerPartition.collectFirst {
                case (tip, _) if tip.topicPartition() == tp => tip.topicId()
              }.getOrElse(Uuid.ZERO_UUID), tp) -> new PartitionResponse(error)
            } ++ results
            responseCallback(mergedResults)
          },
          recordValidationStatsCallback = recordValidationStatsCallback,
          requestLocal = newRequestLocal,
          verificationGuards = verificationGuards
        )
      } else {
        // 如果没有有效数据，直接返回错误响应
        val errorResults = preAppendErrors.map { case (tp, error) =>
          new TopicIdPartition(entriesPerPartition.collectFirst {
            case (tip, _) if tip.topicPartition() == tp => tip.topicId()
          }.getOrElse(Uuid.ZERO_UUID), tp) -> new PartitionResponse(error)
        }
        responseCallback(errorResults)
      }
    }

    // 3. 如果有事务数据，执行验证；否则直接处理
    if (transactionalProducerInfo.nonEmpty && transactionSupportedOperation.isTransactionVerificationRequired()) {
      // 执行事务验证
      txnCoordinator.verifyTransaction(
        transactionalId,
        transactionalProducerInfo.head._1, // producerId
        transactionalProducerInfo.head._2, // producerEpoch
        topicPartitionBatchInfo.toMap,
        requestLocal,
        postVerificationCallback
      )
    } else {
      // 无需验证，直接处理
      postVerificationCallback(requestLocal, (Map.empty, Map.empty))
    }
  }

  /**
   * 将消息记录追加到Leader副本
   * 这是实际执行写入操作的核心方法
   */
  def appendRecords(
    timeout: Long,
    requiredAcks: Short,
    internalTopicsAllowed: Boolean,
    origin: AppendOrigin,
    entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
    responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit,
    recordValidationStatsCallback: Map[TopicIdPartition, RecordValidationStats] => Unit = _ => (),
    requestLocal: RequestLocal = RequestLocal.noCaching,
    verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty
  ): Unit = {
    
    // 1. 验证ACK参数
    if (!isValidRequiredAcks(requiredAcks)) {
      sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback)
      return
    }

    // 2. 写入本地日志
    val localProduceResults = appendRecordsToLeader(
      requiredAcks,
      internalTopicsAllowed,
      origin,
      entriesPerPartition,
      requestLocal,
      defaultActionQueue,
      verificationGuards
    )

    // 3. 构建生产状态
    val produceStatus = buildProducePartitionStatus(localProduceResults)

    // 4. 记录验证统计信息
    recordValidationStatsCallback(localProduceResults.map { case (k, v) =>
      k -> v.info.recordValidationStats
    })

    // 5. 处理延迟生产（等待ISR确认）
    maybeAddDelayedProduce(
      requiredAcks,
      timeout,
      entriesPerPartition,
      localProduceResults,
      produceStatus,
      responseCallback
    )
  }

  /**
   * 将记录写入到Leader副本的本地日志
   */
  private def appendToLocalLog(
    internalTopicsAllowed: Boolean,
    origin: AppendOrigin,
    entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
    requiredAcks: Short,
    requestLocal: RequestLocal,
    verificationGuards: Map[TopicPartition, VerificationGuard]
  ): Map[TopicIdPartition, LogAppendResult] = {
    
    trace(s"追加记录 [$entriesPerPartition] 到本地日志")

    entriesPerPartition.map { case (topicIdPartition, records) =>
      // 更新监控指标
      brokerTopicStats.topicStats(topicIdPartition.topic).totalProduceRequestRate.mark()
      brokerTopicStats.allTopicsStats.totalProduceRequestRate.mark()

      // 拒绝向内部主题写入（如果不允许）
      if (Topic.isInternal(topicIdPartition.topic) && !internalTopicsAllowed) {
        (topicIdPartition, LogAppendResult(
          LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
          Some(new InvalidTopicException(s"不能向内部主题写入 ${topicIdPartition.topic}")),
          hasCustomErrorMessage = false))
      } else {
        try {
          // 获取分区对象并执行写入
          val partition = getPartitionOrException(topicIdPartition)
          val info = partition.appendRecordsToLeader(records, origin, requiredAcks, requestLocal,
            verificationGuards.getOrElse(topicIdPartition.topicPartition(), VerificationGuard.SENTINEL))
          
          val numAppendedMessages = info.numMessages

          // 更新成功写入的统计信息
          brokerTopicStats.topicStats(topicIdPartition.topic).bytesInRate.mark(records.sizeInBytes)
          brokerTopicStats.allTopicsStats.bytesInRate.mark(records.sizeInBytes)
          brokerTopicStats.topicStats(topicIdPartition.topic).messagesInRate.mark(numAppendedMessages)
          brokerTopicStats.allTopicsStats.messagesInRate.mark(numAppendedMessages)

          if (traceEnabled) {
            trace(s"${records.sizeInBytes} 字节写入日志 $topicIdPartition " +
              s"起始偏移量 ${info.firstOffset} 结束偏移量 ${info.lastOffset}")
          }

          (topicIdPartition, LogAppendResult(info, exception = None, hasCustomErrorMessage = false))

        } catch {
          case e @ (_: UnknownTopicOrPartitionException |
                   _: NotLeaderOrFollowerException |
                   _: RecordTooLargeException |
                   _: RecordBatchTooLargeException |
                   _: CorruptRecordException |
                   _: KafkaStorageException) =>
            (topicIdPartition, LogAppendResult(LogAppendInfo.UNKNOWN_LOG_APPEND_INFO, Some(e), hasCustomErrorMessage = false))

          case rve: RecordValidationException =>
            val logAppendInfo = LogAppendInfo.UNKNOWN_LOG_APPEND_INFO
            (topicIdPartition, LogAppendResult(logAppendInfo, Some(rve.invalidException), hasCustomErrorMessage = false))

          case e: Throwable =>
            // 对于未知错误，增加失败计数
            brokerTopicStats.topicStats(topicIdPartition.topic).failedProduceRequestRate.mark()
            brokerTopicStats.allTopicsStats.failedProduceRequestRate.mark()
            (topicIdPartition, LogAppendResult(LogAppendInfo.UNKNOWN_LOG_APPEND_INFO, Some(e), hasCustomErrorMessage = false))
        }
      }
    }
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228

/**
 * SocketServer - 基于NIO的高性能网络服务器
 * 采用Reactor模式处理大量并发连接
 */
class SocketServer(val config: KafkaConfig,
                  val metrics: Metrics,
                  val time: Time,
                  val credentialProvider: CredentialProvider,
                  val apiVersionManager: ApiVersionManager) extends Closeable with Logging {

  // 网络处理器映射 - 每个监听器对应一组处理器
  private val acceptors = mutable.Map[EndPoint, Acceptor]()
  private val processors = mutable.Map[Int, Processor]()
  
  // 请求通道 - 连接网络层和应用层
  private var requestChannel: RequestChannel = _
  private var dataPlaneRequestChannel: RequestChannel = _
  private var controlPlaneRequestChannel: RequestChannel = _

  /**
   * 启动网络服务器
   * 为每个配置的监听器创建Acceptor和Processor
   */
  def startup(
    startProcessingRequests: Boolean = true,
    controlPlaneListener: Option[EndPoint] = None,
    config: KafkaConfig = this.config
  ): Unit = {
    this.synchronized {
      connectionQuotas = new ConnectionQuotas(config, time, metrics)
      
      // 创建请求通道
      createDataPlaneRequestChannel()
      createControlPlaneRequestChannel(controlPlaneListener)

      // 为每个监听器创建Acceptor和Processor
      val sendBufferSize = config.socketSendBufferBytes
      val recvBufferSize = config.socketReceiveBufferBytes
      val maxRequestSize = config.socketRequestMaxBytes

      config.dataPlaneListeners.foreach { endPoint =>
        val listenerName = endPoint.listenerName
        val securityProtocol = endPoint.securityProtocol
        
        // 创建处理器线程池
        val processors = new ArrayBuffer[Processor]()
        for (i <- 0 until config.numNetworkThreads) {
          val processor = new Processor(
            id = nextProcessorId(),
            requestChannel = dataPlaneRequestChannel,
            maxRequestSize = maxRequestSize,
            requestChannelMetrics = dataPlaneRequestChannelMetrics,
            listenerName = listenerName,
            securityProtocol = securityProtocol,
            config = config,
            metrics = metrics,
            time = time,
            credentialProvider = credentialProvider,
            memoryPool = memoryPool,
            logContext = logContext,
            connectionQuotas = connectionQuotas,
            apiVersionManager = apiVersionManager
          )
          processors += processor
          this.processors.put(processor.id, processor)
        }

        // 创建Acceptor
        val acceptor = new Acceptor(
          endPoint = endPoint,
          requestChannel = dataPlaneRequestChannel,
          sendBufferSize = sendBufferSize,
          recvBufferSize = recvBufferSize,
          maxRequestSize = maxRequestSize,
          processors = processors.toBuffer,
          connectionQuotas = connectionQuotas,
          time = time,
          logContext = logContext
        )
        
        acceptors.put(endPoint, acceptor)
        Utils.newThread(s"kafka-socket-acceptor-${listenerName.value}-${endPoint.port}", acceptor, false).start()
        acceptor.awaitStartup()

        // 启动处理器线程
        processors.foreach { processor =>
          Utils.newThread(s"kafka-network-thread-${config.brokerId}-${listenerName.value}-${processor.id}", 
                         processor, false).start()
          processor.awaitStartup()
        }
      }

      info(s"已启动 ${dataPlaneAcceptors.size} 个Acceptor线程，每个都有 ${config.numNetworkThreads} 个处理器线程")
    }
  }

  /**
   * 创建数据平面请求通道
   */
  private def createDataPlaneRequestChannel(): Unit = {
    dataPlaneRequestChannel = new RequestChannel(
      queueSize = config.queuedMaxRequests, 
      numNetworkThreads = config.numNetworkThreads,
      metricNamePrefix = DataPlaneMetricPrefix,
      time = time,
      apiVersionManager = apiVersionManager
    )
  }
}

/**
 * Acceptor - 接受新的客户端连接
 * 运行在独立线程中，将连接分发给Processor处理
 */
private[network] class Acceptor(
  val endPoint: EndPoint,
  val requestChannel: RequestChannel,
  val sendBufferSize: Int,
  val recvBufferSize: Int,
  val maxRequestSize: Int,
  processors: Buffer[Processor],
  connectionQuotas: ConnectionQuotas,
  time: Time,
  logContext: LogContext
) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

  // NIO Selector和ServerSocketChannel
  private val nioSelector = NSelector.open()
  private val serverChannel = openServerSocket(endPoint.host, endPoint.port)
  
  // 处理器分发器 - 轮询分配连接给处理器
  private val processors = mutable.Set[Processor]() ++= processors
  private val processorsIterator = processors.iterator

  /**
   * Acceptor的主循环
   * 接受新连接并分发给处理器
   */
  def run(): Unit = {
    // 注册ServerSocketChannel到Selector
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    startupComplete()
    
    try {
      while (isRunning) {
        try {
          val ready = nioSelector.select(500) // 500ms超时
          if (ready > 0) {
            val keys = nioSelector.selectedKeys()
            val iter = keys.iterator()
            
            while (iter.hasNext && isRunning) {
              try {
                val key = iter.next()
                iter.remove()

                if (key.isAcceptable) {
                  // 接受新连接
                  accept(key).foreach { socketChannel =>
                    // 轮询分配给处理器
                    val processor = synchronized {
                      // 轮询选择处理器
                      currentProcessorIndex = (currentProcessorIndex + 1) % processors.size
                      processors.toSeq(currentProcessorIndex)
                    }
                    
                    // 将新连接分配给处理器
                    processor.accept(socketChannel, maybeBlockingMode = true, maybeBlocked = false)
                  }
                } else {
                  throw new IllegalStateException("未识别的SelectionKey类型")
                }
              } catch {
                case e: Throwable => 
                  error("接受新连接时发生错误", e)
              }
            }
          }
        } catch {
          case e: ControlThrowable => 
            throw e
          case e: Throwable => 
            error("Acceptor线程中发生未预期的错误", e)
        }
      }
    } finally {
      debug("关闭服务器socket和selector")
      CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)
      CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)
      shutdownComplete()
    }
  }

  /**
   * 接受新的客户端连接
   */
  private def accept(key: SelectionKey): Option[SocketChannel] = {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
    val socketChannel = serverSocketChannel.accept()
    
    try {
      // 增加连接计数
      connectionQuotas.inc(socketChannel.socket().getInetAddress)
      
      // 配置socket选项
      socketChannel.configureBlocking(false)
      socketChannel.socket().setTcpNoDelay(true)
      socketChannel.socket().setKeepAlive(true)
      if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) {
        socketChannel.socket().setSendBufferSize(sendBufferSize)
      }
      if (recvBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) {
        socketChannel.socket().setReceiveBufferSize(recvBufferSize)
      }
      
      Some(socketChannel)
    } catch {
      case e: TooManyConnectionsException =>
        info(s"拒绝连接来自 ${socketChannel.socket.getRemoteSocketAddress}：${e.getMessage}")
        close(socketChannel)
        None
      case e: Throwable =>
        error(s"配置接受的socket失败，来自 ${socketChannel.socket.getRemoteSocketAddress}", e)
        close(socketChannel)
        None
    }
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179

/**
 * DelayedProduce - 延迟生产操作
 * 等待ISR副本确认写入后才完成响应
 */
class DelayedProduce(
  timeout: Long,
  produceMetadata: ProduceMetadata,
  replicaManager: ReplicaManager,
  responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit,
  lockOpt: Option[Lock] = None
) extends DelayedOperation(timeout, lockOpt) {

  /**
   * 尝试完成延迟操作
   * 检查所有分区是否都满足了ACK要求
   */
  override def tryComplete(): Boolean = {
    // 检查每个分区的ACK状态
    val (isDone, hasError) = produceMetadata.produceStatus.foldLeft((true, false)) {
      case ((allDone, hasError), (topicIdPartition, status)) =>
        val (isDone, error) = status.acksPending match {
          case 0 => (true, false) // 所有ACK已收到
          case _ => 
            // 检查分区当前状态
            replicaManager.getPartition(topicIdPartition) match {
              case Some(partition) =>
                val (isDone, error) = partition.checkEnoughReplicasReachOffset(status.requiredOffset)
                (isDone, error.isDefined)
              case None =>
                (true, true) // 分区不存在，标记为错误
            }
        }
        (allDone && isDone, hasError || error)
    }

    if (isDone || hasError) {
      // 构建最终响应
      val responseStatus = produceMetadata.produceStatus.map { case (topicIdPartition, status) =>
        if (status.responseStatus.error != Errors.NONE) {
          // 如果已有错误，直接使用
          topicIdPartition -> status.responseStatus
        } else {
          // 检查当前状态
          replicaManager.getPartition(topicIdPartition) match {
            case Some(partition) =>
              val (_, error) = partition.checkEnoughReplicasReachOffset(status.requiredOffset)
              topicIdPartition -> new PartitionResponse(error.getOrElse(Errors.NONE))
            case None =>
              topicIdPartition -> new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)
          }
        }
      }

      // 执行回调
      responseCallback(responseStatus)
      true // 操作完成
    } else {
      false // 操作未完成，继续等待
    }
  }

  /**
   * 操作超时时的处理
   */
  override def onTimeout(): Unit = {
    // 构建超时响应
    val timeoutStatus = produceMetadata.produceStatus.map { case (topicIdPartition, status) =>
      if (status.responseStatus.error != Errors.NONE) {
        topicIdPartition -> status.responseStatus
      } else {
        topicIdPartition -> new PartitionResponse(Errors.REQUEST_TIMED_OUT)
      }
    }

    responseCallback(timeoutStatus)
  }

  /**
   * 当ISR副本状态发生变化时触发完成检查
   */
  override def onExpiration(): Unit = {
    replicaManager.tryCompleteDelayedProduce(new DelayedProduceKey(this.produceMetadata.produceStatus.keys))
  }
}

/**
 * DelayedOperationPurgatory - 延迟操作炼狱
 * 管理所有延迟操作的调度和完成
 */
class DelayedOperationPurgatory[T <: DelayedOperation](
  purgatoryName: String,
  timeWheel: Timer,
  brokerId: Int = 0,
  purgeInterval: Int = 1000,
  reaperEnabled: Boolean = true,
  timerEnabled: Boolean = true
) extends Logging with KafkaMetricsGroup {

  // 监控指标
  private val expirationReaper = new ExpiredOperationReaper(purgatoryName)
  
  // 延迟操作监视器 - 按key组织的延迟操作
  private val watchersForKey = new ConcurrentHashMap[Any, Watchers]()
  
  // 定时器 - 管理操作超时
  private val timeWheelTimer = timeWheel

  /**
   * 检查并完成可以完成的延迟操作
   */
  def checkAndComplete(key: Any): Int = {
    val watchers = inReadLock(removeWatchersLock) { watchersForKey.get(key) }
    val numCompleted = if (watchers == null) {
      0
    } else {
      watchers.tryCompleteWatched()
    }
    debug(s"请求key $key: 完成了 $numCompleted 个延迟操作")
    numCompleted
  }

  /**
   * 尝试完成指定的延迟操作
   */
  def tryCompleteElseWatch(operation: T, watchKeys: Seq[Any]): Boolean = {
    assert(watchKeys.nonEmpty, "至少需要一个watch key")
    
    var isCompletedByMe = operation synchronized operation.tryComplete()
    
    if (isCompletedByMe) {
      isCompletedByMe
    } else {
      // 操作未完成，添加到watch列表
      var watchCreated = false
      for (key <- watchKeys) {
        // 如果操作在此期间被其他线程完成，则停止watch
        if (operation.isCompleted) {
          return false
        }
        watchForOperation(key, operation)
        
        if (!watchCreated) {
          watchCreated = true
          // 添加到时间轮进行超时管理
          timeWheelTimer.add(operation)
        }
      }
      isCompletedByMe
    }
  }

  /**
   * 为指定key添加延迟操作监视
   */
  private def watchForOperation(key: Any, operation: T): Unit = {
    inWriteLock(removeWatchersLock) {
      val watcher = watchersForKey.get(key)
      if (watcher == null) {
        val watchers = new Watchers(key)
        watchers.watch(operation)
        watchersForKey.put(key, watchers)
      } else {
        watcher.watch(operation)
      }
    }
  }

  /**
   * 清理已完成或过期的延迟操作
   */
  def advanceClock(timeout: Long): Unit = {
    timeWheelTimer.advanceClock(timeout)
    
    // 定期清理已完成的操作
    if (reaperEnabled) {
      expirationReaper.advanceClock(timeout)
    }
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198

/**
 * BrokerTopicStats - Broker级别的主题统计信息
 * 用于监控生产和消费的性能指标
 */
class BrokerTopicStats(isRemoteStorageEnabled: Boolean = false) extends KafkaMetricsGroup {
  
  // 全局统计指标
  val allTopicsStats = new TopicStats()
  
  // 每个主题的统计指标
  private val topicStatsMap = new ConcurrentHashMap[String, TopicStats]()

  /**
   * 获取或创建主题的统计信息
   */
  def topicStats(topic: String): TopicStats = {
    topicStatsMap.computeIfAbsent(topic, _ => new TopicStats())
  }

  /**
   * TopicStats - 单个主题的详细统计信息
   */
  class TopicStats extends KafkaMetricsGroup {
    // 消息流量统计
    val messagesInRate = newMeter("MessagesInPerSec", "messages", TimeUnit.SECONDS)
    val bytesInRate = newMeter("BytesInPerSec", "bytes", TimeUnit.SECONDS)
    val bytesOutRate = newMeter("BytesOutPerSec", "bytes", TimeUnit.SECONDS)
    
    // 请求统计
    val totalProduceRequestRate = newMeter("TotalProduceRequestsPerSec", "requests", TimeUnit.SECONDS)
    val totalFetchRequestRate = newMeter("TotalFetchRequestsPerSec", "requests", TimeUnit.SECONDS)
    val failedProduceRequestRate = newMeter("FailedProduceRequestsPerSec", "requests", TimeUnit.SECONDS)
    val failedFetchRequestRate = newMeter("FailedFetchRequestsPerSec", "requests", TimeUnit.SECONDS)
    
    // 远程存储统计（如果启用）
    val remoteReadRequestRate = if (isRemoteStorageEnabled) {
      Some(newMeter("RemoteReadRequestsPerSec", "requests", TimeUnit.SECONDS))
    } else None
    
    val remoteWriteRequestRate = if (isRemoteStorageEnabled) {
      Some(newMeter("RemoteWriteRequestsPerSec", "requests", TimeUnit.SECONDS))
    } else None
    
    // 延迟统计
    val produceMessageConversionsTimeHistogram = newHistogram("ProduceMessageConversionsTimeMs")
    val fetchMessageConversionsTimeHistogram = newHistogram("FetchMessageConversionsTimeMs")
  }

  /**
   * 更新记录转换统计信息
   */
  def updateRecordConversionStats(request: RequestChannel.Request,
                                 tp: TopicPartition,
                                 conversionStats: RecordValidationStats): Unit = {
    if (conversionStats.numRecordsConverted > 0) {
      topicStats(tp.topic).produceMessageConversionsTimeHistogram.update(
        conversionStats.conversionTimeNanos / 1000000)
      allTopicsStats.produceMessageConversionsTimeHistogram.update(
        conversionStats.conversionTimeNanos / 1000000)
    }
  }
}

/**
 * KafkaRequestHandlerPool - 请求处理线程池
 * 负责处理来自网络层的API请求
 */
class KafkaRequestHandlerPool(
  val brokerId: Int,
  val requestChannel: RequestChannel,
  val apis: KafkaApis,
  time: Time,
  numThreads: Int,
  requestHandlerAvgIdleMetricName: String,
  logAndThreadNamePrefix: String
) extends Logging with KafkaMetricsGroup {

  private val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)
  private val threads = new mutable.ArrayBuffer[Thread](numThreads)
  
  // 处理器线程监控指标
  private val aggregateIdleMeter = newMeter(requestHandlerAvgIdleMetricName, "percent", TimeUnit.NANOSECONDS)

  /**
   * 启动请求处理线程池
   */
  def createRequestHandlers(): Unit = {
    for (i <- 0 until numThreads) {
      val handler = new KafkaRequestHandler(
        id = i,
        brokerId = brokerId,
        aggregateIdleMeter = aggregateIdleMeter,
        totalHandlerThreads = numThreads,
        requestChannel = requestChannel,
        apis = apis,
        time = time
      )
      
      runnables += handler
      
      val thread = Utils.newThread(s"$logAndThreadNamePrefix-$i", handler, false)
      threads += thread
      thread.start()
    }
  }

  /**
   * 关闭请求处理线程池
   */
  def shutdown(): Unit = {
    info("正在关闭请求处理器线程池")
    for (handler <- runnables) {
      handler.initiateShutdown()
    }
    for (thread <- threads) {
      thread.interrupt()
    }
    for (thread <- threads) {
      thread.join()
    }
    info("请求处理器线程池已关闭")
  }
}

/**
 * KafkaRequestHandler - 单个请求处理线程
 * 从RequestChannel获取请求并调用KafkaApis处理
 */
class KafkaRequestHandler(
  id: Int,
  brokerId: Int,
  val aggregateIdleMeter: Meter,
  val totalHandlerThreads: Int,
  val requestChannel: RequestChannel,
  apis: KafkaApis,
  time: Time
) extends Runnable with Logging {

  private val shutdownComplete = new CountDownLatch(1)
  @volatile private var stopped = false

  /**
   * 请求处理主循环
   */
  def run(): Unit = {
    val startSelectTime = time.nanoseconds
    
    while (!stopped) {
      // We use a single meter for aggregate idle percentage for the thread pool.
      // Since meter is calculated as total_recorded_value / time_window and
      // time_window is independent of the number of threads, each recorded idle
      // time should be discounted by # threads.
      val endSelectTime = time.nanoseconds
      val idleTime = endSelectTime - startSelectTime
      aggregateIdleMeter.mark(idleTime / totalHandlerThreads)
      
      // 从请求通道获取请求
      val req = requestChannel.receiveRequest(300)
      val startProcessTime = time.nanoseconds
      
      if (req != null) {
        try {
          req.requestDequeueTimeNanos = startProcessTime
          trace(s"Kafka请求处理器 $id 处理请求 ${req.header.apiKey}")
          
          // 调用KafkaApis处理请求
          apis.handle(req, RequestLocal.withThreadConfinedCaching)
        } catch {
          case e: FatalExitError =>
            shutdownComplete.countDown()
            Exit.exit(e.statusCode)
          case e: Throwable => 
            error(s"处理请求时发生异常", e)
        } finally {
          req.releaseBuffer()
        }
      }
      
      val startSelectTime = time.nanoseconds
    }
    shutdownComplete.countDown()
    info(s"Kafka请求处理器 $id 已停止")
  }

  /**
   * 初始化关闭过程
   */
  def initiateShutdown(): Unit = {
    stopped = true
  }

  /**
   * 等待关闭完成
   */
  def awaitShutdown(): Unit = {
    shutdownComplete.await()
  }
}