ableegoldman commented on code in PR #12235:
URL: https://github.com/apache/kafka/pull/12235#discussion_r890725391
##########
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:
Oh man I had no idea you had to close static mocks...not blaming you for
this at all @cadonna but in the future, assume I know nothing about
Mockito/PowerMock and put everything out on the table! lol 🙃
--
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
