...

• We plan to add a new metric 

  kafka.network:name=ControllerRequestQueueSize,type=RequestChannel

  to show the size of the new controller request queue.
• We will add two new metrics 
  kafka.network:name=ControllerNetworkProcessorIdlePercent,type=SocketServer
  kafka.server:name=ControllerRequestHandlerIdlePercent,type=KafkaRequestHandlerPool
  with the former monitoring the idle percentage of pinned controller network thread, and the latter monitoring the idle percentage of pinned controller request handler thread. (Pinned controller threads are explained in details below.)
• The meaning of the existing metric

  kafka.network:name=RequestQueueSize,type=RequestChannel

  will be changed, and it will be used to show the size of the data request queue only, which does not include controller requests.
• The meaning of the two existing metrics
  kafka.network:name=NetworkProcessorAvgIdlePercent,type=SocketServer
  kafka.server:name=RequestHandlerAvgIdlePercent,type=KafkaRequestHandlerPool
  will be changed slightly in that they now only cover the threads for data plane threads, not including the pinned threads for controller requests.
• We plan to add a new config for brokers to specify a dedicated listener for controller connections: the "controller.listener.name" config. A detailed explanation of the config is shown in the "proposed changes" section. When the "controller.listener.name" config is not set explicitly, it will have a default value of CONTROLLER.
• To support the default value of CONTROLLER for the "controller.listener.name" config, the following existing configs that are already deprecated will be removed:
  host
  port
  advertised.host
  advertised.port 
  The reason for removing these configs are explained in following paragraphs. 
• Currently when the "listeners" config and the "advertised.listeners" config are not set explicitly, they rely on the deprecated configs host, port, advertised.host, and advertised.port for default values. In this KIP, we propose to assign default values to "listeners" and "advertised.listeners" so that each default value includes an endpoint for the listener name "CONTROLLER".
When the "listeners" and "advertised.listeners" config are set explicitly, a new listener-to-endpoint entry dedicated to controller connections need to be added.
Also a new entry needs to be added to the "listener.security.protocol.map" to specify the security protocol of the new endpoint.

Proposed Changes

• A new listener-to-endpoint entry dedicated to controller connections need to be added to the "listeners" config and the "advertised.listeners" config.
  Also a new entry needs to be added to the "listener.security.protocol.map" to specify the security protocol of the new endpoint.
  
  

Proposed Changes

In order to eliminate queuing for controller requests, we plan to add dedicated endpoints to brokers for controller connections and dedicated threads for handling controller requests. To explain the proposed change, we first go through how brokers should get the dedicated endpoints through configs, and expose the endpoints to Zookeeper. In order to eliminate queuing for controller requests, we plan to add dedicated endpoints to brokers for controller connections and dedicated threads for handling controller requests. To explain the proposed change, we first go through how brokers should get the dedicated endpoints through configs, and expose the endpoints to Zookeeper. Then we discuss how a controller can learn about the dedicated endpoints exposed by brokers. Finally we describe how controller requests are handled over the dedicated connections.

...

{
	"listener_security_protocol_map": {
		"CONTROLLER": "PLAINTEXT"
        "INTERNAL": "PLAINTEXT",
        "EXTERNAL": "SSL"
    },
    "endpoints": [
		"CONTROLLER://broker1.example.com:9091",
        "INTERNAL://broker1.example.com:9092",
        "EXTERNAL://host1.example.com:9093"
    ],
    "host": "host1.example.com",
    "port": 9092,
9092",
        "jmx_port": -1,EXTERNAL://host1.example.com:9093"
    "timestamp": "1532467569343"],
    "version": 4
}

...

"host": "host1.example.com

...

",
    "port": 9092,
    "jmx_port": -1,
    "timestamp": "1532467569343",
    "version": 4
}

and the "controller.listener.name" config is set to value "CONTROLLER", it will use the corresponding endpoint "CONTROLLER://broker1.example.com:9091" and the security protocol "PLAINTEXT" for connections with this broker.

Whenever the "controller.listener.name" is set, upon broker startup, we will validate its value and make sure it's different from the inter-broker-listener-name value.

If the "controller.listener.name" config is not set, the controller will fall back to the current behavior and use inter-broker-listener-name value to determine controller-to-broker endpoints

Whenever the "controller.listener.name" is set, upon broker startup, we will validate its value and make sure it's different from the inter-broker-listener-name value.

If the "controller.listener.name" config is not set explicitly, the default value "CONTROLLER" will be used. Supporting this default value means when the "advertised.listeners" config is also not set explicitly, there needs to be a default value for it that includes an endpoint for the listener name "CONTROLLER". Supporting a default value for the "advertised.listeners" config means the default values for the deprecated configs "advertised.host" and "advertised.port" will no longer be used, hence we are proposing to remove them in this KIP. As shown in the next section, the same logic applies to the "listeners" config as well, which leads us to propose the removal of the "host" and "port" config as well.

How are controller requests handled over the dedicated connections?

With the dedicated endpoints for controller connections, upon startup a broker will use the "controller.listener.name" to look up the corresponding endpoint in the listeners list for binding. For instance, in the example given above, the broker will derive the dedicated endpoint to be "CONTROLLER://192.1.1.8:9091". Then it will have a new dedicated acceptor that binds to this endpoint, and listens for controller connections. When a connection is received, the socket will be given to a dedicated processor thread (network thread). The dedicated processor thread reads controller requests from the socket and enqueues them to a new dedicated queue for controller requests, whose default capacity is 20 [2].  On the other side of the controller request queue, a dedicated request handler thread will take requests out, and handles them in the same way as being done today. In summary, we are adding a dedicated acceptor, pinning one processor thread, adding a new request queue, and pinning one request handler thread for controller connections and requests. The two new threads are exclusively for requests from the controller and do not handle data plane requests.

If the "controller.listener.name" config is not set, then there is no way to tell the dedicated endpoint for controller, and hence there will be no dedicated acceptor, network processor, or request handler threads. The behavior should be exactly same as the current implementation.

Compatibility, Deprecation, and Migration Plan

...