KAFKA-6504: Fix creation of a sensor to be specific to a metric group so it is not shared #4514
Conversation
…d by copy-paste error
|
Refer to this link for build results (access rights to CI server needed): |
|
@rayokota, we should probably backport this to all branches since 1.0. |
|
@hachikuji , I added a unit test to check the "source-record-active-count" metric. However, the above bug does not cause this test to fail, because the metric name is fine but it is the sensor name that is not. Two metrics ("sink-record-active-count" and "source-record-active-count") were both sharing the sensor with the same name ("sink-record-active-count"). So to trigger the bug, I would need to write a test to invoke both the WorkerSinkTask and the WorkerSourceTask and show that the metrics "source-record-active-count" and "sink-record-active-count" were a combined sum of the both source and sink record active counts when they should not be. At least that is what I believe is going on based on my reading of the code. Do you still want such a test? |
| @@ -509,7 +509,7 @@ public SourceTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) | |||
| pollTime.add(metricGroup.metricName(registry.sourceRecordPollBatchTimeMax), new Max()); | |||
| pollTime.add(metricGroup.metricName(registry.sourceRecordPollBatchTimeAvg), new Avg()); | |||
|
|
|||
| sourceRecordActiveCount = metricGroup.metrics().sensor("sink-record-active-count"); | |||
| sourceRecordActiveCount = metricGroup.metrics().sensor("source-record-active-count"); | |||
There was a problem hiding this comment.
I think the reported bug might actually be the fact that we use metricGroup.metrics().sensor() instead of metricGroup.sensor(). Since we are not going through the MetricGroup, the sensor will not be unique across source tasks. That means when we update this sensor for one source task, we will update all the other source tasks as well. This seems to fit with what was reported in the JIRA. You can verify the bug by adding another metric group to the test case with a different taskId.
Looking over the code, it looks like WorkerSinkTask also suffers from the same problem. We should fix that as well and add test cases which verify sensor uniqueness. It would also be a good idea to add tests which verify that metrics and sensors are removed correctly for a group after closing the metric group.
There was a problem hiding this comment.
You are right, the sensor creation was wrong. I fixed the set up of sensors and added unit tests that verify that the sensors were not unique before the fix, but unique after the fix.
| assertEquals(1900.0, metrics.currentMetricValueAsDouble(group.metricGroup(), "poll-batch-max-time-ms"), 0.001d); | ||
| assertEquals(1450.0, metrics.currentMetricValueAsDouble(group.metricGroup(), "poll-batch-avg-time-ms"), 0.001d); | ||
| assertEquals(33.333, metrics.currentMetricValueAsDouble(group.metricGroup(), "source-record-poll-rate"), 0.001d); | ||
| assertEquals(1000, metrics.currentMetricValueAsDouble(group.metricGroup(), "source-record-poll-total"), 0.001d); | ||
| assertEquals(3.3333, metrics.currentMetricValueAsDouble(group.metricGroup(), "source-record-write-rate"), 0.001d); | ||
| assertEquals(100, metrics.currentMetricValueAsDouble(group.metricGroup(), "source-record-write-total"), 0.001d); | ||
| assertEquals(900.0, metrics.currentMetricValueAsDouble(group.metricGroup(), "source-record-active-count"), 0.001d); | ||
|
|
There was a problem hiding this comment.
Maybe we can close the first group here and verify that the sensors/metrics are no longer registered? A similar check for the sink would be good.
There was a problem hiding this comment.
Ok, I added code to close the first group
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
@rayokota Before I merge, can you change the title and the description to match the updated patch? |
rayokota
changed the title
|
Ok, changed the title and description. Thanks for the review! |
…4514) This change ensures that when sensors are created, they are unique to the metric group associated with the task that created them. Previously the sensors were being shared between task metric groups, causing incorrect metrics. Reviewers: Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>
…4514) This change ensures that when sensors are created, they are unique to the metric group associated with the task that created them. Previously the sensors were being shared between task metric groups, causing incorrect metrics. Reviewers: Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>
This change ensures that when sensors are created, they are specific to a metric group. Previously the sensors were being shared between metric groups, causing incorrect metrics.