splett2 commented on a change in pull request #9386: URL: https://github.com/apache/kafka/pull/9386#discussion_r521549126
########## File path: core/src/main/scala/kafka/network/SocketServer.scala ########## @@ -538,14 +542,20 @@ private[kafka] class Acceptor(val endPoint: EndPoint, val recvBufferSize: Int, brokerId: Int, connectionQuotas: ConnectionQuotas, - metricPrefix: String) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup { + metricPrefix: String, + time: Time) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup { private val nioSelector = NSelector.open() val serverChannel = openServerSocket(endPoint.host, endPoint.port) private val processors = new ArrayBuffer[Processor]() private val processorsStarted = new AtomicBoolean private val blockedPercentMeter = newMeter(s"${metricPrefix}AcceptorBlockedPercent", "blocked time", TimeUnit.NANOSECONDS, Map(ListenerMetricTag -> endPoint.listenerName.value)) + private var currentProcessorIndex = 0 + private[network] case class DelayedCloseSocket(socket: SocketChannel, endThrottleTimeMs: Long) extends Ordered[DelayedCloseSocket] { + override def compare(that: DelayedCloseSocket): Int = endThrottleTimeMs compare that.endThrottleTimeMs + } + private[network] val throttledSockets = new mutable.PriorityQueue[DelayedCloseSocket]() Review comment: I explained elsewhere, but my comment on may have gotten lost in the weeds. for reference: I did not use the java DelayQueue similar to the implementation in ClientQuotaManager because this throttling implementation does not use timeout-based polling or require a synchronized data structure, and there's a bit more boilerplate needed for a DelayQueue to support timeout-based polling. ---------------------------------------------------------------- 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: us...@infra.apache.org