dajac commented on a change in pull request #9386:
URL: https://github.com/apache/kafka/pull/9386#discussion_r503249872
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -1203,14 +1261,28 @@ class ConnectionQuotas(config: KafkaConfig, time: Time,
metrics: Metrics) extend
private val listenerCounts = mutable.Map[ListenerName, Int]()
private[network] val maxConnectionsPerListener = mutable.Map[ListenerName,
ListenerConnectionQuota]()
@volatile private var totalCount = 0
-
+ @volatile private var defaultConnectionRatePerIp =
DynamicConfig.Ip.DefaultConnectionCreationRate
+ private val inactiveSensorExpirationTimeSeconds =
TimeUnit.HOURS.toSeconds(1);
Review comment:
Could we define this constant in a companion object?
##########
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))
Review comment:
nit: Can't we reuse `connectionRateQuota` directly?
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -1394,28 +1554,38 @@ class ConnectionQuotas(config: KafkaConfig, time: Time,
metrics: Metrics) extend
* Creates sensor for tracking the connection creation rate and
corresponding connection rate quota for a given
* listener or broker-wide, if listener is not provided.
* @param quotaLimit connection creation rate quota
- * @param listenerOpt listener name if sensor is for a listener
+ * @param connectionQuotaEntity entity to create the sensor for
*/
- private def createConnectionRateQuotaSensor(quotaLimit: Int, listenerOpt:
Option[String] = None): Sensor = {
- val sensorName = listenerOpt.map(listener =>
s"ConnectionAcceptRate-$listener").getOrElse("ConnectionAcceptRate")
- val sensor = metrics.sensor(sensorName, rateQuotaMetricConfig(quotaLimit))
- sensor.add(connectionRateMetricName(listenerOpt), new Rate, null)
- info(s"Created $sensorName sensor, quotaLimit=$quotaLimit")
- sensor
+ private def getOrCreateConnectionRateQuotaSensor(quotaLimit: Int,
connectionQuotaEntity: ConnectionQuotaEntity): Sensor = {
+ val (sensorName, sensorExpiration) = connectionQuotaEntity match {
+ case BrokerQuotaEntity => ("ConnectionAcceptRate", Long.MaxValue)
+ case listenerEntity: ListenerQuotaEntity =>
(s"ConnectionAcceptRate-${listenerEntity.entityName}", Long.MaxValue)
+ case ipEntity: IpQuotaEntity =>
(s"ConnectionAcceptRate-${ipEntity.entityName}",
inactiveSensorExpirationTimeSeconds)
+ }
Review comment:
Couldn't we rely on the `ConnectionQuotaEntity` trait to get these
informations? That would avoid having to pattern match on the entity to get
them.
##########
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:
- Do we really need to store delayed sockets per IP? It seems that we
just close all the sockets in `throttledSockets` later on so it is not really
necessary. Or have I missed something?
- Similarly, we use a `DelayQueue` for the other quotas. Is there any reason
why we don't use the same data structure here?
##########
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])
+ val endThrottleTimeMs = e.startThrottleTimeMs + e.throttleTimeMs
+ delayQueue += DelayedCloseSocket(socketChannel, endThrottleTimeMs)
+ None
+ }
+ }
+
+ /**
+ * Close sockets for any connections that have been throttled
+ */
+ private def closeThrottledConnections(): Unit = {
+ val timeMs = time.milliseconds
+ val iter = throttledSockets.values.iterator
+ while (iter.hasNext) {
Review comment:
If we were using a shared `DelayQueue` for all IPs, iterating over all
the IPs wouldn't be necessary when there are no throttled connections ready to
be closed yet.
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -1242,7 +1314,56 @@ class ConnectionQuotas(config: KafkaConfig, time: Time,
metrics: Metrics) extend
private[network] def updateBrokerMaxConnectionRate(maxConnectionRate: Int):
Unit = {
// if there is a connection waiting on the rate throttle delay, we will
let it wait the original delay even if
// the rate limit increases, because it is just one connection per
listener and the code is simpler that way
- updateConnectionRateQuota(maxConnectionRate)
+ updateConnectionRateQuota(maxConnectionRate, BrokerQuotaEntity)
+ }
+
+ /**
+ * Update the connection rate quota for a given IP and updates quota configs
for updated IPs.
+ * If an IP is given, metric config will be updated only for the given IP,
otherwise
+ * all metric configs will be checked and updated if required
+ *
+ * @param ip ip to update or default if None
+ * @param maxConnectionRate new connection rate, or resets entity to default
if None
+ */
+ def updateIpConnectionRate(ip: Option[String], maxConnectionRate:
Option[Int]): Unit = {
+ def isIpConnectionRateMetric(metricName: MetricName) = {
+ metricName.name == "connection-accept-rate" &&
Review comment:
Could we use a constant for the name and use it everywhere? Same for the
tag.
##########
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:
nit: Could we use directly the `Admin` client here? We don't use the ZK
admin client in the suite at all so it would be better if we don't use it. I am
not sure if this depends on the pending implementation that you mentioned.
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -1394,28 +1554,38 @@ class ConnectionQuotas(config: KafkaConfig, time: Time,
metrics: Metrics) extend
* Creates sensor for tracking the connection creation rate and
corresponding connection rate quota for a given
* listener or broker-wide, if listener is not provided.
* @param quotaLimit connection creation rate quota
- * @param listenerOpt listener name if sensor is for a listener
+ * @param connectionQuotaEntity entity to create the sensor for
*/
- private def createConnectionRateQuotaSensor(quotaLimit: Int, listenerOpt:
Option[String] = None): Sensor = {
- val sensorName = listenerOpt.map(listener =>
s"ConnectionAcceptRate-$listener").getOrElse("ConnectionAcceptRate")
- val sensor = metrics.sensor(sensorName, rateQuotaMetricConfig(quotaLimit))
- sensor.add(connectionRateMetricName(listenerOpt), new Rate, null)
- info(s"Created $sensorName sensor, quotaLimit=$quotaLimit")
- sensor
+ private def getOrCreateConnectionRateQuotaSensor(quotaLimit: Int,
connectionQuotaEntity: ConnectionQuotaEntity): Sensor = {
+ val (sensorName, sensorExpiration) = connectionQuotaEntity match {
+ case BrokerQuotaEntity => ("ConnectionAcceptRate", Long.MaxValue)
+ case listenerEntity: ListenerQuotaEntity =>
(s"ConnectionAcceptRate-${listenerEntity.entityName}", Long.MaxValue)
+ case ipEntity: IpQuotaEntity =>
(s"ConnectionAcceptRate-${ipEntity.entityName}",
inactiveSensorExpirationTimeSeconds)
+ }
+ sensorAccessor.getOrCreate(
+ sensorName,
+ sensorExpiration,
+ sensor => sensor.add(connectionRateMetricName(connectionQuotaEntity),
new Rate, rateQuotaMetricConfig(quotaLimit))
+ )
}
/**
- * Updates quota configuration for a given listener or broker-wide (if
'listenerOpt' is None)
+ * Updates quota configuration for a given connection quota entity
*/
- private def updateConnectionRateQuota(quotaLimit: Int, listenerOpt:
Option[String] = None): Unit = {
- val metric = metrics.metric(connectionRateMetricName(listenerOpt))
- metric.config(rateQuotaMetricConfig(quotaLimit))
- info(s"Updated ${listenerOpt.getOrElse("broker-wide")} max connection
creation rate to $quotaLimit")
+ private def updateConnectionRateQuota(quotaLimit: Int,
connectionQuotaEntity: ConnectionQuotaEntity): Unit = {
+ val metricOpt =
Option(metrics.metric(connectionRateMetricName(connectionQuotaEntity)))
+ metricOpt.foreach { metric =>
+ metric.config(rateQuotaMetricConfig(quotaLimit))
+ info(s"Updated ${connectionQuotaEntity.entityName} max connection
creation rate to $quotaLimit")
+ }
}
- private def connectionRateMetricName(listenerOpt: Option[String]):
MetricName = {
- val tags = listenerOpt.map(listener => Map("listener" ->
listener)).getOrElse(Map())
- val namePrefix = listenerOpt.map(_ => "").getOrElse("broker-")
+ private def connectionRateMetricName(connectionQuotaEntity:
ConnectionQuotaEntity): MetricName = {
+ val (namePrefix, tags) = connectionQuotaEntity match {
+ case BrokerQuotaEntity => ("broker-", Map.empty[String, String])
+ case listener: ListenerQuotaEntity => ("", Map("listener" ->
listener.entityName))
+ case ip: IpQuotaEntity => ("", Map("ip" -> ip.entityName))
+ }
Review comment:
Same comment as before. Couldn't we rely on the `ConnectionQuotaEntity`
trait here as well?
##########
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:
Instead of recording, checking the quota, and unrecording if the quota
is violated, we may be able to check the quota and record only if the quota is
not violated yet. Your implementation suggests that we want to enforce a strict
quota here. We did so for the controller mutations with the following
implementation:
```
try {
quotaSensor synchronized {
quotaSensor.checkQuotas(timeMs)
quotaSensor.record(permits, timeMs, false)
}
} catch {
case e: QuotaViolationException =>
...
}
```
It seems that this should work here as well. Let me know what you think
about this.
##########
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:
Out of curiosity, have we considered using a "reaper" running in a
separate thread to drain the queue and close throttled connections?
##########
File path: core/src/main/scala/kafka/network/SocketServer.scala
##########
@@ -1242,7 +1314,56 @@ class ConnectionQuotas(config: KafkaConfig, time: Time,
metrics: Metrics) extend
private[network] def updateBrokerMaxConnectionRate(maxConnectionRate: Int):
Unit = {
// if there is a connection waiting on the rate throttle delay, we will
let it wait the original delay even if
// the rate limit increases, because it is just one connection per
listener and the code is simpler that way
- updateConnectionRateQuota(maxConnectionRate)
+ updateConnectionRateQuota(maxConnectionRate, BrokerQuotaEntity)
+ }
+
+ /**
+ * Update the connection rate quota for a given IP and updates quota configs
for updated IPs.
+ * If an IP is given, metric config will be updated only for the given IP,
otherwise
+ * all metric configs will be checked and updated if required
+ *
+ * @param ip ip to update or default if None
+ * @param maxConnectionRate new connection rate, or resets entity to default
if None
+ */
+ def updateIpConnectionRate(ip: Option[String], maxConnectionRate:
Option[Int]): Unit = {
+ def isIpConnectionRateMetric(metricName: MetricName) = {
+ metricName.name == "connection-accept-rate" &&
+ metricName.group == MetricsGroup &&
+ metricName.tags.containsKey("ip")
+ }
+
+ def shouldUpdateQuota(metric: KafkaMetric, quotaLimit: Int) = {
+ quotaLimit != metric.config.quota.bound
+ }
+
+ ip match {
+ case Some(addr) =>
+ val address = InetAddress.getByName(addr)
+ if (maxConnectionRate.isDefined) {
Review comment:
nit: We could use pattern matching here to treat both cases.
----------------------------------------------------------------
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]
