KAFKA-5968: Create/remove request metrics during broker startup/shutdown #3991
Conversation
|
@ijuma Can you review, please? Thank you... |
| } | ||
|
|
||
| def sendShutdownRequest(): Unit = requestQueue.put(ShutdownRequest) | ||
|
|
||
| } | ||
|
|
||
| object RequestMetrics { | ||
| class RequestChannelMetrics { |
There was a problem hiding this comment.
Maybe this can be an inner class in the RequestChannel object called Metrics or MetricsMap.
| metricsMap.put(name, new RequestMetrics(name)) | ||
| } | ||
|
|
||
| def apply(metricName: String) = metricsMap(metricName) | ||
| def requestMetrics(metricName: String) = metricsMap(metricName) |
There was a problem hiding this comment.
Can we not leave it as apply?
| @@ -365,7 +369,7 @@ class RequestMetrics(name: String) extends KafkaMetricsGroup { | |||
| else | |||
| None | |||
|
|
|||
| private val errorMeters = mutable.Map[Errors, ErrorMeter]() | |||
| val errorMeters = mutable.Map[Errors, ErrorMeter]() | |||
There was a problem hiding this comment.
I think it's good to avoid exposing mutable collections publicly. Typically, one can expose the operations required instead.
| removeMetric("TotalTimeMs", tags) | ||
| removeMetric("ResponseSendTimeMs", tags) | ||
| removeMetric("RequestBytes", tags) | ||
| removeMetric("ResponseSendTimeMs", tags) |
| @@ -53,7 +53,8 @@ object RequestChannel extends Logging { | |||
| val context: RequestContext, | |||
| val startTimeNanos: Long, | |||
| memoryPool: MemoryPool, | |||
| @volatile private var buffer: ByteBuffer) extends BaseRequest { | |||
| @volatile private var buffer: ByteBuffer, | |||
| requestChannelMetrics: RequestChannelMetrics) extends BaseRequest { | |||
| @@ -247,6 +249,8 @@ class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMe | |||
| ) | |||
| } | |||
|
|
|||
| def requestMetrics(name: String): RequestMetrics = requestChannelMetrics.requestMetrics(name) | |||
There was a problem hiding this comment.
I removed this method.
|
@ijuma Thanks for the review, updated. |
| } | ||
|
|
||
| class RequestMetrics(name: String) extends KafkaMetricsGroup { | ||
| val tags = Map("request" -> name) | ||
| val requestRate = newMeter("RequestsPerSec", "requests", TimeUnit.SECONDS, tags) | ||
| val requestRate = newMeter(RequestMetrics.RequestsPerSec, "requests", TimeUnit.SECONDS, tags) |
There was a problem hiding this comment.
A common pattern is doing the following at the top of the class:
import RequestMetrics._Then the class has access to the members of the companion object and we don't need to add RequestMetrics everywhere.
There was a problem hiding this comment.
@ijuma Thank you, updated
Replaces the static
RequestMetricsobject with a class so that metrics are created and removed during broker startup and shutdown to avoid metrics tests being affected by metrics left behind by previous tests. Also reinstateskafka.api.MetricsTestwhich was failing frequently earlier due to tests removing the static request metrics.