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 using a DelayQueue.
----------------------------------------------------------------
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
