...

connections.setup.timeout.ms: The configuration controls the maximum amount of time the client will wait for the initial socket connection to be built. If the connection is not built before the timeout elapses the network client will close the socket channel. The default value will be 10 seconds.

Proposed Changes

NetworkClient

1. The new config will be a common client config. The NetworkClient will keep the config

...

I'm proposing to do a lazy socket connection time out. That is, the NetworkClient will only check and disconnect timeout connections in leastLoadedNode(). 

1. Usually, when clients send a request, they will ask the network client to send the request to a specific node. In these cases, the connection.setup.timeout won’t matter too much because the client doesn’t want to try other nodes for that specific request. The request level timeout would be enough. The metadata fetcher fetches the status of the nodes periodically so the clients will reassign the timeout request correspondingly to a different node.
2. Consumer, producer, and AdminClient are all using leastLoadedNode() for metadata fetches, where the connection setup timeout can play an important role. Unlike other requests can refer to the metadata for node condition, the metadata requests can only blindly choose a node for retry in the worst scenario. We want to make sure the client can get the metadata smoothly and as soon as possible. As a result, we need this connection.setup.timeout.
3. Implementing the timeout in NetworkClient.poll() or anywhere else might need an extra iteration of all nodes, which might downgrade the network client performance.
4. .ms as a new property.
5. NetworkClient.poll() will iterate all connecting nodes and disconnect() those connections take more than connection.setup.timeout to finishClients fetch metadata periodically which means leastLoadedNode() will get called frequently. So the timeout checking frequency is guaranteed. 
6. The node providing criteria

...

1. C in the least LoadedNode() will also change

...

1. . Now the criteria should look like below:
   1. Provide the connected node with least number of inflight requests
   2. If no connected node exists, provide the connecting node with the largest index in the cached list of nodes.
   3. If no connected or connecting node exists, provide the disconnected node which respects the reconnect backoff with the least number of failed attempts. Consider the case when we have multiple DISCONNECTED nodes and the time interval between the two provide() invokes is greater than reconnect.backoff.ms. The Provider can provide the same nodes all the time. Thus, the provider should provide the nodes with the least failed attempts among all nodes passing the canConnect() check.

ClusterConnectionStates 

1. Add a new HashSet property ConnectingNodes keeping all the connecting node ids.
2. Will expose a public API that returns the ConnectingNodes mentioned in #1, helping the NetworkClient process the timeout iteration.
3. State transition:
   1. ClusterConnectionStates.connecting() will add the node id to ConnectingNodes
   2. ClusterConnectionStates.ready() will remove the node id to ConnectingNodes
   3. ClusterConnectionStates.disconnected() will remove the node id from ConnectingNodes

Compatibility, Deprecation, and Migration Plan

...

1. Use request.timeout.ms to time out the socket connection at the client level instead of the network client level
   1. request.timeout.ms is at the client/request level. We need one in the NetworkClient level to control the connection states.
   2. The socket connection timeout should be relatively shorter than the request timeout. It's good to have a separate config.
2. Add a new connection state TIMEOUT besides DISCONNECTED, CONNECTING, CHECKING_API_VERSIONS, READY, and AUTHENTICATION_FAILED. 
   1. We don't necessarily need to differentiate the timeout and disconnected states.

3. a lazy socket connection time out. That is, the NetworkClient will only check and disconnect timeout connections in leastLoadedNode(). 

   Pros:
   1. Usually, when clients send a request, they will ask the network client to send the request to a specific node. In these cases, the connection.setup.timeout won’t matter too much because the client doesn’t want to try other nodes for that specific request. The request level timeout would be enough. The metadata fetcher fetches the status of the nodes periodically so the clients will reassign the timeout request correspondingly to a different node.
   2. Consumer, producer, and AdminClient are all using leastLoadedNode() for metadata fetches, where the connection setup timeout can play an important role. Unlike other requests can refer to the metadata for node condition, the metadata requests can only blindly choose a node for retry in the worst scenario. We want to make sure the client can get the metadata smoothly and as soon as possible. As a result, we need this connection.setup.timeout.
   3. Implementing the timeout in NetworkClient.poll() or anywhere else might need an extra iteration of all nodes, which might downgrade the network client performance.
   4. Clients fetch metadata periodically which means leastLoadedNode() will get called frequently. So the timeout checking frequency is guaranteed.

            However, we need a more common and universal timeout for all connections. New scenarios may jump out beside the current metadata fetching scenario