splett2 commented on a change in pull request #9386:
URL: https://github.com/apache/kafka/pull/9386#discussion_r503664955
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -1371,6 +1492,45 @@ class ConnectionQuotas(config: KafkaConfig, time: Time,
metrics: Metrics) extend
}
}
+ /**
+ * To avoid over-recording listener/broker connection rate, we unrecord a
listener or broker connection
+ * if the IP gets throttled later.
+ *
+ * @param listenerName listener to unrecord connection
+ * @param timeMs current time in milliseconds
+ */
+ private def unrecordListenerConnection(listenerName: ListenerName, timeMs:
Long): Unit = {
+ if (!protectedListener(listenerName)) {
+ brokerConnectionRateSensor.record(-1.0, timeMs, false)
+ }
+ maxConnectionsPerListener
+ .get(listenerName)
+ .foreach(_.connectionRateSensor.record(-1.0, timeMs, false))
+ }
+
+ /**
+ * Calculates the delay needed to bring the observed connection creation
rate to the IP limit.
+ * If the connection would cause an IP quota violation, un-record the
connection
+ *
+ * @param address
+ * @param timeMs
+ * @return
+ */
+ private def recordIpConnectionMaybeThrottle(address: InetAddress, timeMs:
Long): Long = {
+ val connectionRateQuota = connectionRateForIp(address)
+ val quotaEnabled = connectionRateQuota !=
DynamicConfig.Ip.UnlimitedConnectionCreationRate
+ if (!quotaEnabled) {
+ return 0
+ }
+ val sensor =
getOrCreateConnectionRateQuotaSensor(connectionRateForIp(address),
IpQuotaEntity(address))
+ val throttleMs = recordAndGetThrottleTimeMs(sensor, timeMs)
+ if (throttleMs > 0) {
+ // unrecord the connection since we won't accept the connection
+ sensor.record(-1.0, timeMs, false)
+ }
Review comment:
I didn't do this because I saw that `checkQuotas` checks whether we are
at the upper bound, not whether recording a new connection would throw an
exception, which I suppose is why you are calling `record` with `checkQuotas =
false`.
I think it makes sense to do here. The actual rate of permitted connections
ends up being 1 higher than otherwise, but I think that's fine.
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -600,43 +607,10 @@ private[kafka] class Acceptor(val endPoint: EndPoint,
serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
startupComplete()
try {
- var currentProcessorIndex = 0
while (isRunning) {
try {
- val ready = nioSelector.select(500)
- if (ready > 0) {
- val keys = nioSelector.selectedKeys()
- val iter = keys.iterator()
- while (iter.hasNext && isRunning) {
- try {
- val key = iter.next
- iter.remove()
-
- if (key.isAcceptable) {
- accept(key).foreach { socketChannel =>
- // Assign the channel to the next processor (using
round-robin) to which the
- // channel can be added without blocking. If
newConnections queue is full on
- // all processors, block until the last one is able to
accept a connection.
- var retriesLeft = synchronized(processors.length)
- var processor: Processor = null
- do {
- retriesLeft -= 1
- processor = synchronized {
- // adjust the index (if necessary) and retrieve the
processor atomically for
- // correct behaviour in case the number of processors
is reduced dynamically
- currentProcessorIndex = currentProcessorIndex %
processors.length
- processors(currentProcessorIndex)
- }
- currentProcessorIndex += 1
- } while (!assignNewConnection(socketChannel, processor,
retriesLeft == 0))
- }
- } else
- throw new IllegalStateException("Unrecognized key state for
acceptor thread.")
- } catch {
- case e: Throwable => error("Error while accepting connection",
e)
- }
- }
- }
+ acceptNewConnections()
+ closeThrottledConnections()
Review comment:
I did consider this. I chose to do the work of closing connections in
the acceptor thread to mirror the `Processor` thread which similarly has a
`openSomeConnections()`, `closeSomeConnections()` loop.
Ditto for `Selector.poll()`
Additionally, as a future extension, we may want to re-try accepting a
connection after its throttle time expires, and adding a new reaper thread
would add additional complexity for that logic.
##########
File path:
core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
##########
@@ -240,6 +256,16 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
s"Admin client connection not closed (initial = $initialConnectionCount,
current = $connectionCount)")
}
+ private def updateIpConnectionRate(ip: Option[String], updatedRate: Int):
Unit = {
+ adminZkClient.changeIpConfig(ip.getOrElse(ConfigEntityName.Default),
Review comment:
yeah, this depends on pending implementation. I planned on using
`adminZkClient` in the interim and then replacing it with `Admin` in the PR
adding `KafkaApis` support.
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -697,6 +714,31 @@ private[kafka] class Acceptor(val endPoint: EndPoint,
info(s"Rejected connection from ${e.ip}, address already has the
configured maximum of ${e.count} connections.")
close(endPoint.listenerName, socketChannel)
None
+ case e: ConnectionThrottledException =>
+ val ip = socketChannel.socket.getInetAddress
+ debug(s"Delaying closing of connection from $ip for
${e.throttleTimeMs} ms")
+ val delayQueue = throttledSockets.computeIfAbsent(ip, _ => new
mutable.Queue[DelayedCloseSocket])
Review comment:
I went with this data structure because throttle end time should be
monotonic for a specific IP but not necessarily for all IPs. By maintaining a
queue per IP, we can get constant time adds/removes for a throttled IP.
The downside of this is that in the case where there are no connections to
unthrottle, `closeThrottledConnections` runs in `O(n)` for number of throttled
IPs.
On the other hand `DelayQueue` uses a heap internally and has `O(log(n))`
for each add/remove, and constant time for checking whether there are IPs to
unthrottle. In scenarios where many connections will get unthrottled at once,
the `DelayQueue` will perform significantly worse, since each
`DelayQueue.dequeue` call will be `O(log(n))`.
I admit this could be a premature optimization, and a delay queue could work
better in practice. Let me know what you think.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
