cmccabe commented on a change in pull request #10931:
URL: https://github.com/apache/kafka/pull/10931#discussion_r663270570
##########
File path: core/src/main/scala/kafka/server/BrokerServer.scala
##########
@@ -331,46 +327,40 @@ class BrokerServer(
new FetchSessionCache(config.maxIncrementalFetchSessionCacheSlots,
KafkaServer.MIN_INCREMENTAL_FETCH_SESSION_EVICTION_MS))
- // Start processing requests once we've caught up on the metadata log,
recovered logs if necessary,
- // and started all services that we previously delayed starting.
- val raftSupport = RaftSupport(forwardingManager, metadataCache,
quotaCache)
+ // Create the request processor objects.
+ val raftSupport = RaftSupport(forwardingManager, metadataCache)
dataPlaneRequestProcessor = new
KafkaApis(socketServer.dataPlaneRequestChannel, raftSupport,
replicaManager, groupCoordinator, transactionCoordinator,
autoTopicCreationManager,
- config.nodeId, config, configRepository, metadataCache, metrics,
authorizer, quotaManagers,
+ config.nodeId, config, metadataCache, metadataCache, metrics,
authorizer, quotaManagers,
fetchManager, brokerTopicStats, clusterId, time, tokenManager,
apiVersionManager)
- dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.nodeId,
socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
- config.numIoThreads,
s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent",
SocketServer.DataPlaneThreadPrefix)
-
- socketServer.controlPlaneRequestChannelOpt.foreach {
controlPlaneRequestChannel =>
- controlPlaneRequestProcessor = new
KafkaApis(controlPlaneRequestChannel, raftSupport,
- replicaManager, groupCoordinator, transactionCoordinator,
autoTopicCreationManager,
- config.nodeId, config, configRepository, metadataCache, metrics,
authorizer, quotaManagers,
- fetchManager, brokerTopicStats, clusterId, time, tokenManager,
apiVersionManager)
+ dataPlaneRequestHandlerPool = new KafkaRequestHandlerPool(config.nodeId,
+ socketServer.dataPlaneRequestChannel, dataPlaneRequestProcessor, time,
+ config.numIoThreads,
s"${SocketServer.DataPlaneMetricPrefix}RequestHandlerAvgIdlePercent",
+ SocketServer.DataPlaneThreadPrefix)
- controlPlaneRequestHandlerPool = new
KafkaRequestHandlerPool(config.nodeId,
socketServer.controlPlaneRequestChannelOpt.get, controlPlaneRequestProcessor,
time,
- 1,
s"${SocketServer.ControlPlaneMetricPrefix}RequestHandlerAvgIdlePercent",
SocketServer.ControlPlaneThreadPrefix)
+ if (socketServer.controlPlaneRequestChannelOpt.isDefined) {
+ throw new RuntimeException("KIP-291 endpoints are not supported when
in KRaft mode.")
Review comment:
I will add "control plane listeners" to the exception text.
Incidentally, this should be unreachable because of validation in
`KafkaConfig.scala`, but it never hurts to check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
