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

gitlw · 2018-10-11T18:33:00Z

Separating controller connections and requests from the data plane (KIP-291)

Tested that there are no dedicated control plane endpoints or threads when control.plane.listener.name is not set.
Tested that the controller requests are handled by the control plane threads when
control.plane.listener.name is set

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…ata plane (KIP-291)

gitlw · 2018-10-11T22:11:36Z

retest this please

gitlw · 2018-10-16T21:48:18Z

@junrao @jjkoshy @MayureshGharat Can you please take a look at this PR? Thanks

MayureshGharat · 2018-10-31T18:01:19Z

@gitlw Thanks for the patch. On it now.

MayureshGharat · 2018-11-01T00:17:53Z

core/src/main/scala/kafka/network/SocketServer.scala

-  val requestChannel = new RequestChannel(maxQueuedRequests)
-  private val processors = new ConcurrentHashMap[Int, Processor]()
+  val dataRequestChannel = new RequestChannel(maxQueuedRequests, RequestChannel.RequestQueueSizeMetric)
+  var controlPlaneRequestChannel: RequestChannel = null


Suggested change

var controlPlaneRequestChannel: RequestChannel = null

val controlPlaneRequestChannel: RequestChannel = if (config.controlPlaneListenerName.isDefined) {

controlPlaneRequestChannel = new RequestChannel(20, RequestChannel.ControlPlaneRequestQueueSizeMetric)

} else {

null

}

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

MayureshGharat · 2018-11-01T18:56:20Z

core/src/main/scala/kafka/network/SocketServer.scala

+  }
+  private val dataProcessors = new ConcurrentHashMap[Int, Processor]()
+  // 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]()


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()

MayureshGharat · 2018-11-01T19:03:01Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -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)


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)

}

MayureshGharat · 2018-11-01T21:15:56Z

core/src/main/scala/kafka/network/SocketServer.scala

-      processors.asScala.values.foreach(_.shutdown())
-      requestChannel.clear()
+      dataProcessors.asScala.values.foreach(_.shutdown())
+      controlPlaneProcessors.asScala.values.foreach(_.shutdown())


with the above suggestion, this would be :

Suggested change

controlPlaneProcessors.asScala.values.foreach(_.shutdown())

controlPlaneProcessorOpt.map(_.shutdown())

MayureshGharat · 2018-11-01T21:43:24Z

core/src/main/scala/kafka/network/SocketServer.scala

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


With the above suggestion, this would be :

Suggested change

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

addDataProcessors(acceptor, endpoint, newNumNetworkThreads - oldNumNetworkThreads)

MayureshGharat · 2018-11-01T21:45:39Z

core/src/main/scala/kafka/network/SocketServer.scala

+      controlPlaneProcessors.asScala.values.foreach(_.shutdown())
+      dataRequestChannel.clear()
+      if (controlPlaneRequestChannel != null)
+        controlPlaneRequestChannel.clear()


This would become :

Suggested change

controlPlaneRequestChannel.clear()

controlPlaneRequestChannelOpt.map(_.clear())

MayureshGharat · 2018-11-01T21:57:05Z

core/src/main/scala/kafka/network/SocketServer.scala

-      requestChannel.shutdown()
+      dataRequestChannel.shutdown()
+      if (controlPlaneRequestChannel != null )
+        controlPlaneRequestChannel.shutdown()


with the above suggestion :

Suggested change

controlPlaneRequestChannel.shutdown()

controlPlaneRequestChannelOpt.map(_.shutdown())

MayureshGharat · 2018-11-01T21:58:00Z

core/src/main/scala/kafka/network/SocketServer.scala

+    if (dataPlaneListenersAdded.nonEmpty)
+      createAcceptorAndProcessors(config.numNetworkThreads, dataPlaneListenersAdded, dataRequestChannel, dataProcessors, false)
+    if (controlPlaneListenersAdded.nonEmpty)
+      createAcceptorAndProcessors(1, controlPlaneListenersAdded, controlPlaneRequestChannel, controlPlaneProcessors, true)


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

MayureshGharat · 2018-11-01T22:28:42Z

core/src/main/scala/kafka/server/KafkaServer.scala

  var authorizer: Option[Authorizer] = None
  var socketServer: SocketServer = null
-  var requestHandlerPool: KafkaRequestHandlerPool = null
+  var dataRequestHandlerPool: KafkaRequestHandlerPool = null


should we rename this to "dataPlaneRequestHandlerPool" to be consistent with "controlPlaneRequestHandlerPool" ?

lindong28

Thanks for the patch @gitlw. Left some initial comments.

lindong28 · 2018-11-12T20:44:52Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -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)


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

lindong28 · 2018-11-12T20:45:39Z

core/src/main/scala/kafka/network/RequestChannel.scala

@@ -40,6 +40,7 @@ object RequestChannel extends Logging {
  private val requestLogger = Logger("kafka.request.logger")

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


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).

lindong28 · 2018-11-12T20:56:07Z

core/src/main/scala/kafka/network/SocketServer.scala

-  val requestChannel = new RequestChannel(maxQueuedRequests)
-  private val processors = new ConcurrentHashMap[Int, Processor]()
+  val dataRequestChannel = new RequestChannel(maxQueuedRequests, RequestChannel.RequestQueueSizeMetric)
+  var controlPlaneRequestChannel: RequestChannel = null


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))

lindong28 · 2018-11-12T21:01:03Z

core/src/main/scala/kafka/network/SocketServer.scala

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


nits: can we name it dataPlaneProcessors

lindong28 · 2018-11-12T21:04:59Z

core/src/main/scala/kafka/network/SocketServer.scala

-      createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)
+      createAcceptorAndProcessors(config.numNetworkThreads, config.dataListeners, dataRequestChannel, dataProcessors, false)
+      if (config.controlPlaneListener.isDefined)
+        createAcceptorAndProcessors(1, Seq(config.controlPlaneListener.get), controlPlaneRequestChannel, controlPlaneProcessors, true)


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.

lindong28 · 2018-11-12T21:47:17Z

core/src/main/scala/kafka/network/SocketServer.scala

      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())}


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.

lindong28 · 2018-11-12T22:11:44Z

core/src/main/scala/kafka/server/KafkaRequestHandler.scala

@@ -40,6 +40,8 @@ class KafkaRequestHandler(id: Int,
                          apis: KafkaApis,
                          time: Time) extends Runnable with Logging {
  this.logIdent = "[Kafka Request Handler " + id + " on Broker " + brokerId + "], "
+  // the requestsHandled counter is defined only for testing
+  var requestsHandled: Long = 0L


We generally don't want to add state/variable specifically for unit test, particularly if there is no existing code which does this.

Usually we can do unit test using mockit to verify that something is triggered. Can you see if this is doable? Also, maybe we can just read the newly defined metrics value to verify that the newly defined processor/acceptor has processed some requests.

lindong28 · 2018-11-12T22:12:35Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -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


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?

lindong28 · 2018-11-12T22:14:21Z

core/src/main/scala/kafka/server/KafkaRequestHandler.scala

@@ -96,21 +99,23 @@ class KafkaRequestHandlerPool(val brokerId: Int,
                              val requestChannel: RequestChannel,
                              val apis: KafkaApis,
                              time: Time,
-                              numThreads: Int) extends Logging with KafkaMetricsGroup {
+                              numThreads: Int,
+                              requestHandlerAvgIdleMetric: String,


nits: requestHandlerAvgIdleMetric => requestHandlerAvgIdleMetricName seems better.

lindong28 · 2018-11-12T22:31:14Z

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -113,10 +113,12 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP

  val brokerState: BrokerState = new BrokerState

-  var apis: KafkaApis = null
+  var dataPlaneApis: KafkaApis = null
+  var controlPlaneApis: KafkaApis = null


KafkaApis contains the logic to handle request and itself should contain no state (other than those state in its constructor parameters). So it seems weird to have two KafkaApis. Currently KafkaApis only uses requestChannel with together with RequestChannel.Request. One reasonable approach is to put the requestChannel in RequestChannel.Request similar to the existing RequestChannel.Request.processor. So that we no longer need two KafkaApis instance. I am not sure this is the best alternative approach though.

gitlw · 2018-11-16T20:16:44Z

Closing this PR as @MayureshGharat is taking over the ownership in #5921

KAFKA-4453: Separating controller connections and requests from the d…

a915c3a

…ata plane (KIP-291)

Cleaning all metrics in setUp of InactiveControlPlaneTest

5d0ee9c

MayureshGharat reviewed Nov 1, 2018

View reviewed changes

lindong28 self-assigned this Nov 12, 2018

lindong28 reviewed Nov 12, 2018

View reviewed changes

gitlw closed this Nov 16, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

gitlw commented Oct 11, 2018

gitlw commented Oct 11, 2018

gitlw commented Oct 16, 2018

MayureshGharat commented Oct 31, 2018

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018 •

edited

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018 •

edited

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018

MayureshGharat Nov 1, 2018

lindong28 left a comment

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

lindong28 Nov 12, 2018

gitlw commented Nov 16, 2018

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

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

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

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

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

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

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

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

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

Conversation

gitlw commented Oct 11, 2018

Committer Checklist (excluded from commit message)

gitlw commented Oct 11, 2018

gitlw commented Oct 16, 2018

MayureshGharat commented Oct 31, 2018

Choose a reason for hiding this comment

MayureshGharat Nov 1, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MayureshGharat Nov 1, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lindong28 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gitlw commented Nov 16, 2018

MayureshGharat Nov 1, 2018 •

edited

MayureshGharat Nov 1, 2018 •

edited