dajac commented on a change in pull request #9386:
URL: https://github.com/apache/kafka/pull/9386#discussion_r525888208
##########
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:
Thanks for the clarification. I have missed that previous comment.
----------------------------------------------------------------
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]
