You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2022/06/23 07:45:27 UTC
[kafka] branch trunk updated: MINOR: Fix static mock usage in ProcessorNodeMetricsTest (#12323)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 269277f73b MINOR: Fix static mock usage in ProcessorNodeMetricsTest (#12323)
269277f73b is described below
commit 269277f73bd3c18c378ea3fa6ccc7eb79b914d15
Author: Bruno Cadonna <ca...@apache.org>
AuthorDate: Thu Jun 23 09:45:22 2022 +0200
MINOR: Fix static mock usage in ProcessorNodeMetricsTest (#12323)
Before this PR the calls to StreamsMetricsImpl.addInvocationRateAndCountToSensor()
were just calls and not a verification on the mock. This miss happened
during the switch from EasyMock to Mockito.
Reviewers: John Roesler <vv...@apache.org>, Guozhang Wang <wa...@gmail.com>
---
.../metrics/ProcessorNodeMetricsTest.java | 196 +++++++++++----------
1 file changed, 102 insertions(+), 94 deletions(-)
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java
index 0ae1a99e6f..738d83683e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java
@@ -18,15 +18,18 @@ package org.apache.kafka.streams.processor.internals.metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
-import org.junit.Test;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import org.junit.Test;
+import org.mockito.MockedStatic;
import java.util.Collections;
import java.util.Map;
import java.util.function.Supplier;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TASK_LEVEL_GROUP;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.PROCESSOR_NODE_LEVEL_GROUP;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -50,19 +53,15 @@ public class ProcessorNodeMetricsTest {
final String descriptionOfCount = "The total number of emitted records from the suppression buffer";
final String descriptionOfRate = "The average number of emitted records from the suppression buffer per second";
when(streamsMetrics.nodeLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, metricNamePrefix, RecordingLevel.DEBUG))
- .thenReturn(expectedSensor);
+ .thenReturn(expectedSensor);
when(streamsMetrics.nodeLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID)).thenReturn(tagMap);
- StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedSensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- tagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- );
- verifySensor(
- () -> ProcessorNodeMetrics.suppressionEmitSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics));
+ getAndVerifySensor(
+ () -> ProcessorNodeMetrics.suppressionEmitSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics),
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
+ );
}
@Test
@@ -71,18 +70,14 @@ public class ProcessorNodeMetricsTest {
final String descriptionOfCount = "The total number of skipped idempotent updates";
final String descriptionOfRate = "The average number of skipped idempotent updates per second";
when(streamsMetrics.nodeLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, metricNamePrefix, RecordingLevel.DEBUG))
- .thenReturn(expectedSensor);
+ .thenReturn(expectedSensor);
when(streamsMetrics.nodeLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID)).thenReturn(tagMap);
- StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedSensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- tagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- );
- verifySensor(
- () -> ProcessorNodeMetrics.skippedIdempotentUpdatesSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics)
+
+ getAndVerifySensor(
+ () -> ProcessorNodeMetrics.skippedIdempotentUpdatesSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics),
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
);
}
@@ -92,26 +87,35 @@ public class ProcessorNodeMetricsTest {
final String descriptionOfCount = "The total number of calls to process";
final String descriptionOfRate = "The average number of calls to process per second";
when(streamsMetrics.taskLevelSensor(THREAD_ID, TASK_ID, metricNamePrefix, RecordingLevel.DEBUG))
- .thenReturn(expectedParentSensor);
+ .thenReturn(expectedParentSensor);
when(streamsMetrics.taskLevelTagMap(THREAD_ID, TASK_ID))
- .thenReturn(parentTagMap);
- StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedParentSensor,
- StreamsMetricsImpl.TASK_LEVEL_GROUP,
- parentTagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- );
- setUpThroughputSensor(
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount,
- RecordingLevel.DEBUG,
- expectedParentSensor
- );
-
- verifySensor(() -> ProcessorNodeMetrics.processAtSourceSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics));
+ .thenReturn(parentTagMap);
+ setUpThroughputSensor(metricNamePrefix, RecordingLevel.DEBUG, expectedParentSensor);
+
+ try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
+ final Sensor sensor = ProcessorNodeMetrics.processAtSourceSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics);
+ streamsMetricsStaticMock.verify(
+ () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
+ expectedSensor,
+ PROCESSOR_NODE_LEVEL_GROUP,
+ tagMap,
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
+ )
+ );
+ streamsMetricsStaticMock.verify(
+ () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
+ expectedParentSensor,
+ TASK_LEVEL_GROUP,
+ parentTagMap,
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
+ )
+ );
+ assertThat(sensor, is(expectedSensor));
+ }
}
@Test
@@ -119,65 +123,69 @@ public class ProcessorNodeMetricsTest {
final String metricNamePrefix = "forward";
final String descriptionOfCount = "The total number of calls to forward";
final String descriptionOfRate = "The average number of calls to forward per second";
- setUpThroughputParentSensor(
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- );
- setUpThroughputSensor(
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount,
- RecordingLevel.DEBUG,
- expectedParentSensor
- );
-
- verifySensor(() -> ProcessorNodeMetrics.forwardSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics));
- }
-
- private void setUpThroughputParentSensor(final String metricNamePrefix,
- final String descriptionOfRate,
- final String descriptionOfCount) {
when(streamsMetrics.taskLevelSensor(THREAD_ID, TASK_ID, metricNamePrefix, RecordingLevel.DEBUG))
- .thenReturn(expectedParentSensor);
+ .thenReturn(expectedParentSensor);
when(streamsMetrics.nodeLevelTagMap(THREAD_ID, TASK_ID, StreamsMetricsImpl.ROLLUP_VALUE))
- .thenReturn(parentTagMap);
- StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedParentSensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- parentTagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- );
+ .thenReturn(parentTagMap);
+ setUpThroughputSensor(metricNamePrefix, RecordingLevel.DEBUG, expectedParentSensor);
+
+ try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
+ final Sensor sensor = ProcessorNodeMetrics.forwardSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics);
+ streamsMetricsStaticMock.verify(
+ () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
+ expectedSensor,
+ PROCESSOR_NODE_LEVEL_GROUP,
+ tagMap,
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
+ )
+ );
+ streamsMetricsStaticMock.verify(
+ () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
+ expectedParentSensor,
+ PROCESSOR_NODE_LEVEL_GROUP,
+ parentTagMap,
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
+ )
+ );
+ assertThat(sensor, is(expectedSensor));
+ }
}
private void setUpThroughputSensor(final String metricNamePrefix,
- final String descriptionOfRate,
- final String descriptionOfCount,
- final RecordingLevel recordingLevel,
- final Sensor... parentSensors) {
+ final RecordingLevel recordingLevel,
+ final Sensor... parentSensors) {
when(streamsMetrics.nodeLevelSensor(
- THREAD_ID,
- TASK_ID,
- PROCESSOR_NODE_ID,
- metricNamePrefix,
- recordingLevel,
- parentSensors
+ THREAD_ID,
+ TASK_ID,
+ PROCESSOR_NODE_ID,
+ metricNamePrefix,
+ recordingLevel,
+ parentSensors
)).thenReturn(expectedSensor);
when(streamsMetrics.nodeLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID)).thenReturn(tagMap);
- StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedSensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- tagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- );
}
- private void verifySensor(final Supplier<Sensor> sensorSupplier) {
- final Sensor sensor = sensorSupplier.get();
- assertThat(sensor, is(expectedSensor));
+ private void getAndVerifySensor(final Supplier<Sensor> sensorSupplier,
+ final String metricNamePrefix,
+ final String descriptionOfRate,
+ final String descriptionOfCount) {
+ try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
+ final Sensor sensor = sensorSupplier.get();
+ streamsMetricsStaticMock.verify(
+ () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
+ expectedSensor,
+ PROCESSOR_NODE_LEVEL_GROUP,
+ tagMap,
+ metricNamePrefix,
+ descriptionOfRate,
+ descriptionOfCount
+ )
+ );
+ assertThat(sensor, is(expectedSensor));
+ }
}
}
\ No newline at end of file