chia7712 commented on code in PR #13285:
URL: https://github.com/apache/kafka/pull/13285#discussion_r1123116645
##########
core/src/main/scala/kafka/network/SocketServer.scala:
##########
@@ -115,22 +115,23 @@ class SocketServer(val config: KafkaConfig,
private var stopped = false
// Socket server metrics
- newGauge(s"${DataPlaneAcceptor.MetricPrefix}NetworkProcessorAvgIdlePercent",
() => SocketServer.this.synchronized {
- val dataPlaneProcessors = dataPlaneAcceptors.asScala.values.flatMap(a =>
a.processors)
+ newGauge(s"${DataPlaneAcceptor.MetricPrefix}NetworkProcessorAvgIdlePercent",
() => {
+ val dataPlaneProcessors = dataPlaneAcceptors.values.asScala.flatMap(a =>
a.processors.asScala)
+ // copy to an immutable array to avoid concurrency issue when calculating
average
val ioWaitRatioMetricNames = dataPlaneProcessors.map { p =>
metrics.metricName("io-wait-ratio", MetricsGroup, p.metricTags)
- }
- if (dataPlaneProcessors.isEmpty) {
+ }.toArray
Review Comment:
> To mitigate it, we can store the size in an int before this and then, we
don't have to convert ioWaitRatioMetricNames to an array.
the `ioWaitRatioMetricNames` might reflect modifications of `processors`, so
the actual size of processors may get changed when we calculate the average.
For example, we cache the size = 5 but the `processors` could be increased to 6
when we summarize all `processors`.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
