Skip to content

KAFKA-4453: Separating controller connections and requests from the data plane (KIP-291) #5783

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gitlw wants to merge 2 commits into from
Closed

KAFKA-4453: Separating controller connections and requests from the data plane (KIP-291) #5783

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a915c3a
KAFKA-4453: Separating controller connections and requests from the d…
gitlw Oct 11, 2018
5d0ee9c
Cleaning all metrics in setUp of InactiveControlPlaneTest
gitlw Oct 12, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions core/src/main/scala/kafka/controller/ControllerChannelManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,14 +108,16 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
private def addNewBroker(broker: Broker) {
val messageQueue = new LinkedBlockingQueue[QueueItem]
debug(s"Controller ${config.brokerId} trying to connect to broker ${broker.id}")
val brokerNode = broker.node(config.interBrokerListenerName)
val controlPlaneListenerName = config.controlPlaneListenerName.getOrElse(config.interBrokerListenerName)
val controlPlaneSecurityProtocol = config.controlPlaneSecurityProtocol.getOrElse(config.interBrokerSecurityProtocol)
val brokerNode = broker.node(controlPlaneListenerName)
val logContext = new LogContext(s"[Controller id=${config.brokerId}, targetBrokerId=${brokerNode.idString}] ")
val networkClient = {
val channelBuilder = ChannelBuilders.clientChannelBuilder(
config.interBrokerSecurityProtocol,
controlPlaneSecurityProtocol,
JaasContext.Type.SERVER,
config,
config.interBrokerListenerName,
controlPlaneListenerName,
config.saslMechanismInterBrokerProtocol,
config.saslInterBrokerHandshakeRequestEnable
)
Expand Down
5 changes: 3 additions & 2 deletions core/src/main/scala/kafka/network/RequestChannel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ object RequestChannel extends Logging {
private val requestLogger = Logger("kafka.request.logger")

val RequestQueueSizeMetric = "RequestQueueSize"
val ControlPlaneRequestQueueSizeMetric = "ControlPlaneRequestQueueSize"

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nits: could we also rename variable RequestQueueSizeMetric to DataPlaneRequestQueueSizeMetric to be consistent with the variable name ControlPlaneRequestQueueSizeMetric? This could also reduce confusion for the variable requestQueueSizeMetric used in class RequestChannel(val queueSize: Int, val requestQueueSizeMetric: String).

val ResponseQueueSizeMetric = "ResponseQueueSize"
val ProcessorMetricTag = "processor"

Expand Down Expand Up @@ -272,13 +273,13 @@ object RequestChannel extends Logging {
}
}

class RequestChannel(val queueSize: Int) extends KafkaMetricsGroup {
class RequestChannel(val queueSize: Int, val requestQueueSizeMetric: String) extends KafkaMetricsGroup {
import RequestChannel._
val metrics = new RequestChannel.Metrics
private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
private val processors = new ConcurrentHashMap[Int, Processor]()

newGauge(RequestQueueSizeMetric, new Gauge[Int] {
newGauge(requestQueueSizeMetric, new Gauge[Int] {
def value = requestQueue.size
})

Expand Down
102 changes: 72 additions & 30 deletions core/src/main/scala/kafka/network/SocketServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,8 +65,14 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
private val memoryPoolDepletedTimeMetricName = metrics.metricName("MemoryPoolDepletedTimeTotal", "socket-server-metrics")
memoryPoolSensor.add(new Meter(TimeUnit.MILLISECONDS, memoryPoolDepletedPercentMetricName, memoryPoolDepletedTimeMetricName))
private val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE
val requestChannel = new RequestChannel(maxQueuedRequests)
private val processors = new ConcurrentHashMap[Int, Processor]()
val dataRequestChannel = new RequestChannel(maxQueuedRequests, RequestChannel.RequestQueueSizeMetric)

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nits: to be consistent with variable name controlPlaneRequestChannel, would we name this parameter dataPlaneRequestChannel?

var controlPlaneRequestChannel: RequestChannel = null

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
var controlPlaneRequestChannel: RequestChannel = null
val controlPlaneRequestChannel: RequestChannel = if (config.controlPlaneListenerName.isDefined) {
controlPlaneRequestChannel = new RequestChannel(20, RequestChannel.ControlPlaneRequestQueueSizeMetric)
} else {
null
}

@MayureshGharat MayureshGharat Nov 1, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually how about :

Suggested change
var controlPlaneRequestChannel: RequestChannel = null
private var controlPlaneRequestChannelOpt: Option[RequestChannel] = None

We can then create the controlPlaneRequestChannel lazily when we call startup

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In general in scala we prefer to use val and use None to indicate null. Can we do the following:

var controlPlaneRequestChannel: Option[RequestChannel] = config.controlPlaneListenerName.map(_ => new RequestChannel(20, RequestChannel.ControlPlaneRequestQueueSizeMetric))

if (config.controlPlaneListenerName.isDefined) {
controlPlaneRequestChannel = new RequestChannel(20, RequestChannel.ControlPlaneRequestQueueSizeMetric)
}
private val dataProcessors = new ConcurrentHashMap[Int, Processor]()

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nits: can we name it dataPlaneProcessors

// there should be only one controller processor, however we use a map to store it so that we can reuse the logic for data processors
private[network] val controlPlaneProcessors = new ConcurrentHashMap[Int, Processor]()

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of having a map, which might indicate that we might have multiple processors for the controlPlane, do you think we can do :

Suggested change
private[network] val controlPlaneProcessors = new ConcurrentHashMap[Int, Processor]()
private[network] var controlPlaneProcessorOpt : Option[Processor] = None

We can lazily create controlPlaneProcessor when we call startup()

private var nextProcessorId = 0

private[network] val acceptors = new ConcurrentHashMap[EndPoint, Acceptor]()
Expand All @@ -88,25 +94,35 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
def startup(startupProcessors: Boolean = true) {
this.synchronized {
connectionQuotas = new ConnectionQuotas(config.maxConnectionsPerIp, config.maxConnectionsPerIpOverrides)
createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)
createAcceptorAndProcessors(config.numNetworkThreads, config.dataListeners, dataRequestChannel, dataProcessors, false)

@MayureshGharat MayureshGharat Nov 1, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Was thinking if we should have 2 explicit functions :

  • createDataPlaneAcceptorsAndProcessors(processorsPerListener : Int, endpoints: Seq[EndPoint])

  • createControlPlaneAcceptorAndProcessor (controlEndPointOpt : Option[EndPoint])

We can probably do something like :

Suggested change
createAcceptorAndProcessors(config.numNetworkThreads, config.dataListeners, dataRequestChannel, dataProcessors, false)
createDataPlaneAcceptorsAndProcessors(config.numNetworkThreads, config.dataListeners)
createControlPlaneAcceptorAndProcessor(config.controlPlaneListener)

The two functions can look something like this :

private def createDataPlaneAcceptorsAndProcessors(processorsPerListener: Int,
endpoints: Seq[EndPoint]) : Unit = synchronized {
endpoints.foreach { endpoint =>
val acceptor = createAcceptorWithProcessors()
addDataProcessors(acceptor, endpoint, processorsPerListener, dataRequestChannel, dataProcessors)
KafkaThread.nonDaemon(s"data-plane-kafka-socket-acceptor-${endpoint.listenerName}-${endpoint.securityProtocol}-${endpoint.port}", acceptor).start()
acceptor.awaitStartup()
acceptors.put(endpoint, acceptor)
}
}

private def createControlPlaneAcceptorAndProcessor (controlEndPointOpt: Option[EndPoint]) : Unit = synchronized {
if (controlEndPointOpt.isDefined) {
val controlPlaneEndPoint = controlEndPointOpt.get
controlPlaneRequestChannelOpt = Some(new RequestChannel(20, RequestChannel.ControlPlaneRequestQueueSizeMetric))
val controlPlaneAcceptor = createAcceptor(controlPlaneEndPoint)
val controlPlaneProcessor = newProcessor(nextProcessorId, controlPlaneRequestChannelOpt.get, connectionQuotas, controlPlaneEndPoint.listenerName, controlPlaneEndPoint.securityProtocol, memoryPool, isControlPlane = true)
controlPlaneRequestChannelOpt.get.addProcessor(controlPlaneProcessor)
val listenerProcessors = new ArrayBufferProcessor
listenerProcessors += controlPlaneProceesor
controlPlaneAcceptor.addProcessors(listenerProcessors)
KafkaThread.nonDaemon(s"control-plane-kafka-socket-acceptor-$controlPlaneListenerName-$controlPlaneSecurityProtocol-${controlPlaneEndPoint.port}", controlPlaneAcceptor).start()
controlPlaneAcceptor.awaitStartup()
acceptors.put(controlPlaneEndPoint, controlPlaneAcceptor)
}
}

private def createAcceptor(endpoint: EndPoint): Acceptor = synchronized {
val sendBufferSize = config.socketSendBufferBytes
val recvBufferSize = config.socketReceiveBufferBytes
val brokerId = config.brokerId
new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)
}

private def addDataProcessors(acceptor: Acceptor, endpoint: EndPoint,
newProcessorsPerListener: Int): Unit = synchronized {
val listenerName = endpoint.listenerName
val securityProtocol = endpoint.securityProtocol
val listenerProcessors = new ArrayBufferProcessor

for (_ <- 0 until newProcessorsPerListener) {
  val processor = newProcessor(nextProcessorId, dataRequestChannel, connectionQuotas, listenerName, securityProtocol, memoryPool, isControlPlane = false)
  listenerProcessors += processor
  requestChannel.addProcessor(processor)
  nextProcessorId += 1
}
listenerProcessors.foreach(p => dataProcessors.put(p.id, p))
acceptor.addProcessors(listenerProcessors)

}

if (config.controlPlaneListener.isDefined)
createAcceptorAndProcessors(1, Seq(config.controlPlaneListener.get), controlPlaneRequestChannel, controlPlaneProcessors, true)

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we simplify the code here to createAcceptorAndProcessors(1, config.controlPlaneListener.toSeq, controlPlaneRequestChannel, controlPlaneProcessors, true) without checking config.controlPlaneListener.isDefined? createAcceptorAndProcessors will do nothing if endpoints is an empty sequence.


if (startupProcessors) {
startProcessors()
}
}

newGauge("NetworkProcessorAvgIdlePercent",
new Gauge[Double] {
def createNetworkProcessorAvgIdlePercentMetric(metric: String, processors: java.util.Map[Int, Processor]): Unit = {
newGauge(metric,
new Gauge[Double] {

def value = SocketServer.this.synchronized {
val ioWaitRatioMetricNames = processors.values.asScala.map { p =>
metrics.metricName("io-wait-ratio", "socket-server-metrics", p.metricTags)
def value = SocketServer.this.synchronized {
val ioWaitRatioMetricNames = processors.values.asScala.map { p =>
metrics.metricName("io-wait-ratio", "socket-server-metrics", p.metricTags)
}
ioWaitRatioMetricNames.map { metricName =>
Option(metrics.metric(metricName)).fold(0.0)(m => Math.min(m.metricValue.asInstanceOf[Double], 1.0))
}.sum / processors.size
}
ioWaitRatioMetricNames.map { metricName =>
Option(metrics.metric(metricName)).fold(0.0)(m => Math.min(m.metricValue.asInstanceOf[Double], 1.0))
}.sum / processors.size
}
}
)
)
}

createNetworkProcessorAvgIdlePercentMetric("NetworkProcessorAvgIdlePercent", dataProcessors)
if (config.controlPlaneListener.isDefined)

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is probably simpler to always have the metric defined for user to read. The value of the metric can be n/a similar to the design in KIP-386. By doing this, regardless of the value of broker config, user can always expect to see all possible metrics and only have to deal with the metric value.

And if we follow this design, we do not need to check config.controlPlaneListener.isDefined here. createNetworkProcessorAvgIdlePercentMetric() can handle the case where processors.size == 0.

createNetworkProcessorAvgIdlePercentMetric("ControlPlaneNetworkProcessorIdlePercent", controlPlaneProcessors)

newGauge("MemoryPoolAvailable",
new Gauge[Long] {
def value = memoryPool.availableMemory()
Expand All @@ -133,7 +149,10 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
private def endpoints = config.listeners.map(l => l.listenerName -> l).toMap

private def createAcceptorAndProcessors(processorsPerListener: Int,
endpoints: Seq[EndPoint]): Unit = synchronized {
endpoints: Seq[EndPoint],
requestChannel: RequestChannel,
processorCollector: java.util.Map[Int, Processor],
isControlPlane: Boolean): Unit = synchronized {

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The value of isControlPlane is only used to determine the prefix of the thread name. And the logic of if (processor.isControlPlane) "control" else "data") duplicated three times in the code.

Would it be simpler and more readable to just pass the threadName from the beginning?

Also, could we following the existing code style where each variable in the new line are aligned?


val sendBufferSize = config.socketSendBufferBytes
val recvBufferSize = config.socketReceiveBufferBytes
Expand All @@ -144,25 +163,27 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
val securityProtocol = endpoint.securityProtocol

val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)
addProcessors(acceptor, endpoint, processorsPerListener)
KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()
addProcessors(acceptor, endpoint, processorsPerListener, requestChannel, processorCollector, isControlPlane)
KafkaThread.nonDaemon((if (isControlPlane) "control" else "data") + s"-plane-kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()
acceptor.awaitStartup()
acceptors.put(endpoint, acceptor)
}
}

private def addProcessors(acceptor: Acceptor, endpoint: EndPoint, newProcessorsPerListener: Int): Unit = synchronized {
private def addProcessors(acceptor: Acceptor, endpoint: EndPoint,

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nits: Could we follow the existing code style such that, if we want to break the parameters into multiple lines, each parameter has its own line and these lines are aligned with the first parameter?

newProcessorsPerListener: Int, requestChannel: RequestChannel, processorCollector: java.util.Map[Int, Processor],
isControlPlane: Boolean): Unit = synchronized {
val listenerName = endpoint.listenerName
val securityProtocol = endpoint.securityProtocol
val listenerProcessors = new ArrayBuffer[Processor]()

for (_ <- 0 until newProcessorsPerListener) {
val processor = newProcessor(nextProcessorId, connectionQuotas, listenerName, securityProtocol, memoryPool)
val processor = newProcessor(nextProcessorId, requestChannel, connectionQuotas, listenerName, securityProtocol, memoryPool, isControlPlane)
listenerProcessors += processor
requestChannel.addProcessor(processor)
nextProcessorId += 1
}
listenerProcessors.foreach(p => processors.put(p.id, p))
listenerProcessors.foreach(p => processorCollector.put(p.id, p))
acceptor.addProcessors(listenerProcessors)
}

Expand All @@ -173,21 +194,30 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
info("Stopping socket server request processors")
this.synchronized {
acceptors.asScala.values.foreach(_.shutdown())
processors.asScala.values.foreach(_.shutdown())
requestChannel.clear()
dataProcessors.asScala.values.foreach(_.shutdown())
controlPlaneProcessors.asScala.values.foreach(_.shutdown())

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

with the above suggestion, this would be :

Suggested change
controlPlaneProcessors.asScala.values.foreach(_.shutdown())
controlPlaneProcessorOpt.map(_.shutdown())

dataRequestChannel.clear()
if (controlPlaneRequestChannel != null)
controlPlaneRequestChannel.clear()

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would become :

Suggested change
controlPlaneRequestChannel.clear()
controlPlaneRequestChannelOpt.map(_.clear())

stoppedProcessingRequests = true
}
info("Stopped socket server request processors")
}

def resizeThreadPool(oldNumNetworkThreads: Int, newNumNetworkThreads: Int): Unit = synchronized {
info(s"Resizing network thread pool size for each listener from $oldNumNetworkThreads to $newNumNetworkThreads")
val dataAcceptors = config.controlPlaneListenerName match {
case Some(controlPlaneListenerName) =>
acceptors.asScala.filter{ case (endpoint, _) => !endpoint.listenerName.value().equals(controlPlaneListenerName.value())}

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In general it is preferred to identify type based on e.g. enum, or use separate variables (e.g. dataPlaneAccetor vs. controlPlaneAcceptor), rather than by matching the string value. Could you see if there is a way to improve it.

case None => acceptors.asScala
}

if (newNumNetworkThreads > oldNumNetworkThreads) {
acceptors.asScala.foreach { case (endpoint, acceptor) =>
addProcessors(acceptor, endpoint, newNumNetworkThreads - oldNumNetworkThreads)
dataAcceptors.foreach { case (endpoint, acceptor) =>
addProcessors(acceptor, endpoint, newNumNetworkThreads - oldNumNetworkThreads, dataRequestChannel, dataProcessors, false)

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With the above suggestion, this would be :

Suggested change
addProcessors(acceptor, endpoint, newNumNetworkThreads - oldNumNetworkThreads, dataRequestChannel, dataProcessors, false)
addDataProcessors(acceptor, endpoint, newNumNetworkThreads - oldNumNetworkThreads)

}
} else if (newNumNetworkThreads < oldNumNetworkThreads)
acceptors.asScala.values.foreach(_.removeProcessors(oldNumNetworkThreads - newNumNetworkThreads, requestChannel))
dataAcceptors.values.foreach(_.removeProcessors(oldNumNetworkThreads - newNumNetworkThreads, dataRequestChannel))
}

/**
Expand All @@ -199,7 +229,9 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
this.synchronized {
if (!stoppedProcessingRequests)
stopProcessingRequests()
requestChannel.shutdown()
dataRequestChannel.shutdown()
if (controlPlaneRequestChannel != null )
controlPlaneRequestChannel.shutdown()

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

with the above suggestion :

Suggested change
controlPlaneRequestChannel.shutdown()
controlPlaneRequestChannelOpt.map(_.shutdown())

}
info("Shutdown completed")
}
Expand All @@ -215,7 +247,12 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time

def addListeners(listenersAdded: Seq[EndPoint]): Unit = synchronized {
info(s"Adding listeners for endpoints $listenersAdded")
createAcceptorAndProcessors(config.numNetworkThreads, listenersAdded)
val (controlPlaneListenersAdded, dataPlaneListenersAdded) = listenersAdded.partition { endpoint =>
config.controlPlaneListenerName.isDefined && endpoint.listenerName.value() == config.controlPlaneListenerName.get }
if (dataPlaneListenersAdded.nonEmpty)
createAcceptorAndProcessors(config.numNetworkThreads, dataPlaneListenersAdded, dataRequestChannel, dataProcessors, false)
if (controlPlaneListenersAdded.nonEmpty)
createAcceptorAndProcessors(1, controlPlaneListenersAdded, controlPlaneRequestChannel, controlPlaneProcessors, true)

@MayureshGharat MayureshGharat Nov 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we have a check that a control plane listener is already defined, since there should be only one controlPlaneListener, no?

startProcessors()
}

Expand All @@ -237,9 +274,10 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
}

/* `protected` for test usage */
protected[network] def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,
securityProtocol: SecurityProtocol, memoryPool: MemoryPool): Processor = {
protected[network] def newProcessor(id: Int, requestChannel: RequestChannel, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,
securityProtocol: SecurityProtocol, memoryPool: MemoryPool, isControlPlane: Boolean): Processor = {
new Processor(id,
isControlPlane,
time,
config.socketRequestMaxBytes,
requestChannel,
Expand All @@ -261,7 +299,7 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
Option(connectionQuotas).fold(0)(_.get(address))

/* For test usage */
private[network] def processor(index: Int): Processor = processors.get(index)
private[network] def processor(index: Int): Processor = dataProcessors.get(index)

}

Expand Down Expand Up @@ -355,7 +393,7 @@ private[kafka] class Acceptor(val endPoint: EndPoint,

private def startProcessors(processors: Seq[Processor]): Unit = synchronized {
processors.foreach { processor =>
KafkaThread.nonDaemon(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
KafkaThread.nonDaemon((if (processor.isControlPlane) "control" else "data") + s"-plane-kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
processor).start()
}
}
Expand Down Expand Up @@ -498,6 +536,7 @@ private[kafka] object Processor {
* each of which has its own selector
*/
private[kafka] class Processor(val id: Int,
val isControlPlane: Boolean,
time: Time,
maxRequestSize: Int,
requestChannel: RequestChannel,
Expand Down Expand Up @@ -528,6 +567,8 @@ private[kafka] class Processor(val id: Int,
override def toString: String = s"$localHost:$localPort-$remoteHost:$remotePort-$index"
}

// the receivesProcessed counter is defined only for testing
private[network] var receivesProcessed: Long = 0L

@lindong28 lindong28 Nov 12, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We generally don't want to define variable and state just for unit tests. Can you find a way to test the code without adding such variables?

private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()
private val inflightResponses = mutable.Map[String, RequestChannel.Response]()
private val responseQueue = new LinkedBlockingDeque[RequestChannel.Response]()
Expand Down Expand Up @@ -707,6 +748,7 @@ private[kafka] class Processor(val id: Int,
val req = new RequestChannel.Request(processor = id, context = context,
startTimeNanos = time.nanoseconds, memoryPool, receive.payload, requestChannel.metrics)
requestChannel.sendRequest(req)
receivesProcessed += 1
selector.mute(connectionId)
handleChannelMuteEvent(connectionId, ChannelMuteEvent.REQUEST_RECEIVED)
case None =>
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -640,7 +640,7 @@ class DynamicThreadPool(server: KafkaServer) extends BrokerReconfigurable {

override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
if (newConfig.numIoThreads != oldConfig.numIoThreads)
server.requestHandlerPool.resizeThreadPool(newConfig.numIoThreads)
server.dataRequestHandlerPool.resizeThreadPool(newConfig.numIoThreads)
if (newConfig.numNetworkThreads != oldConfig.numNetworkThreads)
server.socketServer.resizeThreadPool(oldConfig.numNetworkThreads, newConfig.numNetworkThreads)
if (newConfig.numReplicaFetchers != oldConfig.numReplicaFetchers)
Expand Down
Loading