KAFKA-7730: Limit number of active connections per listener in brokers (KIP-402)

[rajinisivaram]

Adds a new listener config max.connections to limit the number of active connections on each listener. The config may be prefixed with listener prefix. This limit may be dynamically reconfigured without restarting the broker.

This is one of the PRs for KIP-402 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-402%3A+Improve+fairness+in+SocketServer+processors). Note that this is currently built on top of PR #6022

Committer Checklist (excluded from commit message)

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

[gwenshap]

I'm having trouble understanding the way the different connection limits work in relation to each other...

• if I exceeded the per-IP limit, we'll throw an exception, log an "info" and close the connection
• if I exceed the per-broker limit, we'll wait for an open slot, but on the next processor iteration, we'll close a connection and get the open slot, so this is a relatively short block.
• if I exceed the per-listener limit but not the per-broker limit, we'll wait for an open slot, but since the maxConnectionsExceeded method only checks the per-broker limit, we may potentially wait for quite a while until a client disconnects from our socket.

I'm not clear if the three different behaviors are intentional. Especially the third one which could be intentional but could also be a bug :)

Can you explain the behavior a bit? I've read the KIP, but I didn't find this detail.

[rajinisivaram]

[~gwenshap] The behaviours are intentional - not necessarily correct though :-)

1. If we exceed per-IP limit, we disconnect any new connections. We need to accept the connection first to detect the source IP, so this is a connect followed by a disconnect. This is the existing behaviour.

2. If we exceed the overall broker limit, we handle inter-broker connections differently from other connections. We never block inter-broker connections due to overall limit - i.e a runaway client on another listener cannot bring the broker down by preventing inter-broker connections from being made. We allow inter-broker connection to succeed regardless of the limit and then close the least recently used connection even though it may not have reached the idle timeout.

3. If we exceed overall broker limit and a connection arrives on a non-inter-broker listener or exceed the listener limit on any listener, then we don't accept new connections until some connection is closed normally or due to hitting the idle timeout. If this is a short connection storm, then new connections are blocked to protect the broker, but as connections are closed, new connections will be accepted. Clients will just see a short delay in connection establishment. If we have a long period of time with a large number of active connections hitting broker or listener limit, then we may block new connections for a long time. In this case clients will timeout and attempt to reconnect. We can hit this scenario if brokers are configured with lower limits than required by client applications. We have metrics that show active connections as well as the amount of time blocked, so brokers can be reconfigured if required.

[gwenshap]

1. Got it.

2. This is the "protectedListener" part, yes? It is smart and awesome.

3. I think I got it. We accept a connection, we see that we now exceed the broker limit and we block. Which means that we are not accepting any other connection until we have a free slot. One thing that is maybe worth documenting: The listening socket is created with a backlog of 50 (nio default), so clients will start getting errors if more of those requests accumulate. I'd probably want to monitor the backlog of the TCP socket to make sure it remains reasonable. Or, maybe we want to expose an option for users to override this? I guess a good value depends on the connection storms you anticipate?

4. I'm concerned that:

  def maxConnectionsExceeded(listenerName: ListenerName): Boolean = {
    totalCount > brokerMaxConnections && !protectedListener(listenerName)
  }

Doesn't take per-listener limits into account. Doesn't it mean that if we have a limit on SSL listener but not on the broker, we'll never close excess connections on that listener? I feel like I'm missing something.

[rajinisivaram]

3. We didn't add a configuration option for server backlog queue size since it is very hard to come up with a good value. As you said it would depend on the nature of connection storms that are hard to predict.

4. Let's say we have two listeners, a SASL_SSL listener for external connections and an SSL listener for internal connections. We configure listener limits for both and an overall broker limit to protect the broker.
   4a) If we hit the listener limit on the SSL listener, we will block the next connection until a connection is closed. We would expect the limit to be reasonably high for the SSL listener, so we would never expect to block with well-behaved brokers.
   4b) If we hit the listener limit on the SASL_SSL listener, we will block the next connection until a connection is closed. We would expect the limit to be set according to the connections expected from client applications. You can monitor blocking times and increase limits if required.
   4c) You can set a broker limit that is less than the sum of listener limits. The higher listener limits improve utilization of connection slots and the lower broker limit protects the broker. We allow both the SASL_SSL and SSL listeners to use up to their individual listener limit, but beyond that they are blocked until a connection on that listener is closed. If the total exceeds the broker limit without hitting the listener limits, then we allow SSL connection to succeed and close the least recently used SASL_SSL connection. We dont allow any more SASL_SSL connections until a connection is closed normally (disconnected or idle timeout).

[gwenshap]

The asymmetric behavior where we accept connections up to the listener limits and disconnect connections based on the broker limits (but done separately on each processor for each listener) makes the interactions really complex to reason about.

If you have a broker limit, an internal (protected) listener and 2 external listeners (not totally unreasonable scenario), thats a lot of combinations to consider.
You can have one listener with blocked connections (because the listener limit is exceeded), while another listener connection getting closed (because broker limit is exceeded) but still accepting new connections (because its own limit wasn't exceeded yet). It just seems difficult to reason about the possible scenarios and figure out the right limits to set and the expected behavior.

I think I'm missing something about the reasoning for the different behaviors.
Why not close connections on a listener, if that listener limit is exceeded? Why wait for the broker limit to exceed?
And the inverse, why close connections if the broker limit is exceeded? Why not have connections wait for other connections to "naturally" close, the way you currently do for listeners?

[rajinisivaram]

@gwenshap Agree that connection limit handling is complex and the fact that we are handling different limits in different ways makes implementation and debugging more complex. Not sure if it makes usage more complex though since the limits serve different purposes. Listener limits require the user to understand the number of applications connecting to each listener and connections per-broker for those. While broker limits should be configured based on the total number of connections and total memory reserved for those connections (e.g. SSL buffers).

1. Why not close connections on a listener, if that listener limit is exceeded? Why wait for the broker limit to exceed?
   If we close old connections to accommodate new connections on the broker, old ones would retry and we would go into a loop of connection plus expensive authentications, overloading the broker. By preventing new connections, we would still have connection retry, but without expensive authentications.

2. And the inverse, why close connections if the broker limit is exceeded? Why not have connections wait for other connections to "naturally" close, the way you currently do for listeners?
   We ensure that application connections cannot bring a broker down by blocking inter-broker connections by closing external connections to make space for inter-broker connections. When we complete each iteration, the total number of connections is guaranteed to be within the broker limit. If we allow new connections without closing existing ones, total number of connections can be above broker limit for unbounded periods of time. And users will need to configure broker limit to be (broker-limit-based-on-resource-allocation minus expected-inter-broker-connections).

[gwenshap]

Thanks for the patient explanation. I think it is reasonable to want both high limit to protect the broker that is aggressively enforced, and also "QoS" limits to allocate connections to listeners and allow for something like "fast lane" listeners for important apps.

[gwenshap]

LGTM. Code looks right, tests are awesome. Great work :)

Left a comment regarding the docs: I think the feature requires more explanation, especially on how broker and listener limits are different.

Feel free to merge after you add explanation, or wait for me to merge tomorrow morning :)

[gwenshap]

Maybe add an explanation of how max.connections is enforced vs how listener.max.connections, and how we expect max.connections to be relatively high and set based on broker capacity, while listener.max.connections is to allocate connections to applications and should be set based on application requirements.

It took me few extra explanations to get it, so I think a bit more of explanation here will help.

[rajinisivaram]

@gwenshap Thanks for the review, I have updated the doc.

[gwenshap]

LGTM. Thank you.