ableegoldman commented on code in PR #12235:
URL: https://github.com/apache/kafka/pull/12235#discussion_r888782190
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java:
##########
@@ -114,6 +116,63 @@ public void shouldGetProcessAtSourceSensor() {
verifySensor(() ->
ProcessorNodeMetrics.processAtSourceSensor(THREAD_ID, TASK_ID,
PROCESSOR_NODE_ID, streamsMetrics));
}
+ @Test
+ public void shouldGetRecordsAndBytesConsumedSensor() {
+ final String recordsMetricNamePrefix = "records-consumed";
+ final String bytesMetricNamePrefix = "bytes-consumed";
+ final String descriptionOfRecordsTotal = "The total number of records
consumed from this topic";
+ final String descriptionOfBytesTotal = "The total number of bytes
consumed from this topic";
+ when(streamsMetrics.nodeLevelSensor(THREAD_ID, TASK_ID,
PROCESSOR_NODE_ID, recordsMetricNamePrefix, RecordingLevel.INFO))
+ .thenReturn(expectedSensor);
+ when(streamsMetrics.nodeLevelSensor(THREAD_ID, TASK_ID,
PROCESSOR_NODE_ID, bytesMetricNamePrefix, RecordingLevel.INFO))
+ .thenReturn(expectedSensor);
+ when(streamsMetrics.nodeLevelTagMap(THREAD_ID, TASK_ID,
PROCESSOR_NODE_ID)).thenReturn(tagMap);
+
+ final Map<String, String> consumedTagMap = new HashMap<>(tagMap);
+ consumedTagMap.put("topic", TOPIC_NAME);
+ StreamsMetricsImpl.addTotalCountAndSumMetricsToSensor(
Review Comment:
Ah, thanks for the explanation. I updated the test as suggested (now in
`TopicMetricsTest`)
--
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
