You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/12 11:00:08 UTC
[GitHub] [kafka] cadonna commented on a diff in pull request #12459: KAFKA-13036: Replace EasyMock and PowerMock with Mockito for RocksDBMetricsRecorderTest
cadonna commented on code in PR #12459:
URL: https://github.com/apache/kafka/pull/12459#discussion_r944226241
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -70,105 +73,133 @@ public class RocksDBMetricsRecorderTest {
private final Statistics statisticsToAdd2 = mock(Statistics.class);
private final Statistics statisticsToAdd3 = mock(Statistics.class);
- private final Sensor bytesWrittenToDatabaseSensor = createMock(Sensor.class);
- private final Sensor bytesReadFromDatabaseSensor = createMock(Sensor.class);
- private final Sensor memtableBytesFlushedSensor = createMock(Sensor.class);
- private final Sensor memtableHitRatioSensor = createMock(Sensor.class);
- private final Sensor memtableAvgFlushTimeSensor = createMock(Sensor.class);
- private final Sensor memtableMinFlushTimeSensor = createMock(Sensor.class);
- private final Sensor memtableMaxFlushTimeSensor = createMock(Sensor.class);
- private final Sensor writeStallDurationSensor = createMock(Sensor.class);
- private final Sensor blockCacheDataHitRatioSensor = createMock(Sensor.class);
- private final Sensor blockCacheIndexHitRatioSensor = createMock(Sensor.class);
- private final Sensor blockCacheFilterHitRatioSensor = createMock(Sensor.class);
- private final Sensor bytesReadDuringCompactionSensor = createMock(Sensor.class);
- private final Sensor bytesWrittenDuringCompactionSensor = createMock(Sensor.class);
- private final Sensor compactionTimeAvgSensor = createMock(Sensor.class);
- private final Sensor compactionTimeMinSensor = createMock(Sensor.class);
- private final Sensor compactionTimeMaxSensor = createMock(Sensor.class);
- private final Sensor numberOfOpenFilesSensor = createMock(Sensor.class);
- private final Sensor numberOfFileErrorsSensor = createMock(Sensor.class);
-
- private final StreamsMetricsImpl streamsMetrics = niceMock(StreamsMetricsImpl.class);
+ private final Sensor bytesWrittenToDatabaseSensor = mock(Sensor.class);
+ private final Sensor bytesReadFromDatabaseSensor = mock(Sensor.class);
+ private final Sensor memtableBytesFlushedSensor = mock(Sensor.class);
+ private final Sensor memtableHitRatioSensor = mock(Sensor.class);
+ private final Sensor memtableAvgFlushTimeSensor = mock(Sensor.class);
+ private final Sensor memtableMinFlushTimeSensor = mock(Sensor.class);
+ private final Sensor memtableMaxFlushTimeSensor = mock(Sensor.class);
+ private final Sensor writeStallDurationSensor = mock(Sensor.class);
+ private final Sensor blockCacheDataHitRatioSensor = mock(Sensor.class);
+ private final Sensor blockCacheIndexHitRatioSensor = mock(Sensor.class);
+ private final Sensor blockCacheFilterHitRatioSensor = mock(Sensor.class);
+ private final Sensor bytesReadDuringCompactionSensor = mock(Sensor.class);
+ private final Sensor bytesWrittenDuringCompactionSensor = mock(Sensor.class);
+ private final Sensor compactionTimeAvgSensor = mock(Sensor.class);
+ private final Sensor compactionTimeMinSensor = mock(Sensor.class);
+ private final Sensor compactionTimeMaxSensor = mock(Sensor.class);
+ private final Sensor numberOfOpenFilesSensor = mock(Sensor.class);
+ private final Sensor numberOfFileErrorsSensor = mock(Sensor.class);
+
+ private final StreamsMetricsImpl streamsMetrics = mock(StreamsMetricsImpl.class);
private final RocksDBMetricsRecordingTrigger recordingTrigger = mock(RocksDBMetricsRecordingTrigger.class);
private final RocksDBMetricsRecorder recorder = new RocksDBMetricsRecorder(METRICS_SCOPE, STORE_NAME);
+ private MockedStatic<RocksDBMetrics> dbMetrics;
+
@Before
public void setUp() {
- setUpMetricsStubMock();
- expect(streamsMetrics.rocksDBMetricsRecordingTrigger()).andStubReturn(recordingTrigger);
- replay(streamsMetrics);
+ setUpMetricsMock();
+ when(streamsMetrics.rocksDBMetricsRecordingTrigger()).thenReturn(recordingTrigger);
recorder.init(streamsMetrics, TASK_ID1);
}
+ @After
+ public void cleanUpMocks() {
+ dbMetrics.close();
+ }
+
@Test
public void shouldInitMetricsRecorder() {
- setUpMetricsMock();
-
- recorder.init(streamsMetrics, TASK_ID1);
-
- verify(RocksDBMetrics.class);
+ dbMetrics.verify(() -> RocksDBMetrics.bytesWrittenToDatabaseSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.bytesReadFromDatabaseSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.memtableBytesFlushedSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.memtableHitRatioSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.memtableAvgFlushTimeSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.memtableMinFlushTimeSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.memtableMaxFlushTimeSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.writeStallDurationSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.blockCacheDataHitRatioSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.blockCacheIndexHitRatioSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.blockCacheFilterHitRatioSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.bytesWrittenDuringCompactionSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.bytesReadDuringCompactionSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.compactionTimeAvgSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.compactionTimeMinSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.compactionTimeMaxSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.numberOfOpenFilesSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.numberOfFileErrorsSensor(any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumImmutableMemTableMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addCurSizeActiveMemTable(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addCurSizeAllMemTables(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addSizeAllMemTables(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumEntriesActiveMemTableMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumEntriesImmMemTablesMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumDeletesActiveMemTableMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumDeletesImmMemTablesMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addMemTableFlushPending(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumRunningFlushesMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addCompactionPendingMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumRunningCompactionsMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addEstimatePendingCompactionBytesMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addTotalSstFilesSizeMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addLiveSstFilesSizeMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addNumLiveVersionMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addBlockCacheCapacityMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addBlockCacheUsageMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addBlockCachePinnedUsageMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addEstimateNumKeysMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addEstimateTableReadersMemMetric(any(), any(), any()));
+ dbMetrics.verify(() -> RocksDBMetrics.addBackgroundErrorsMetric(any(), any(), any()));
assertThat(recorder.taskId(), is(TASK_ID1));
}
@Test
public void shouldThrowIfMetricRecorderIsReInitialisedWithDifferentTask() {
- setUpMetricsStubMock();
- recorder.init(streamsMetrics, TASK_ID1);
-
assertThrows(
- IllegalStateException.class,
- () -> recorder.init(streamsMetrics, TASK_ID2)
+ IllegalStateException.class,
+ () -> recorder.init(streamsMetrics, TASK_ID2)
Review Comment:
nit: The indentation was actually correct before. We use 4 spaces to keep the code a bit more compact, the code is not 100% consistent with rule, though. Same applies to other indentations in this file.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
Review Comment:
The use of `reset()` is discouraged (see https://javadoc.io/doc/org.mockito/mockito-core/latest/org/mockito/Mockito.html#reset(T...)).
You can equally verify that the method `recordingTrigger.addMetricsRecorder(recorder)` is called twice.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
Review Comment:
To avoid `reset()` here, you can split this test method into two:
- in one test method you add two value providers, remove only one, and verify that `removeMetricsRecorder` is never called and
- in the second test method you add two value providers, remove both, and verify that `removeMetricsRecorder` is called only once.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- verify(recordingTrigger);
+ verify(recordingTrigger, never()).removeMetricsRecorder(recorder);
- reset(recordingTrigger);
- recordingTrigger.removeMetricsRecorder(recorder);
- replay(recordingTrigger);
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_2);
- verify(recordingTrigger);
+ verify(recordingTrigger).removeMetricsRecorder(recorder);
}
@Test
public void shouldThrowIfValueProvidersToRemoveNotFound() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
assertThrows(
- IllegalStateException.class,
- () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
+ IllegalStateException.class,
+ () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
);
}
@Test
public void shouldRecordStatisticsBasedMetrics() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(statisticsToAdd1);
- reset(statisticsToAdd2);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(1L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(2L);
- bytesWrittenToDatabaseSensor.record(1 + 2, 0L);
- replay(bytesWrittenToDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(3L);
- bytesReadFromDatabaseSensor.record(2 + 3, 0L);
- replay(bytesReadFromDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(4L);
- memtableBytesFlushedSensor.record(3 + 4, 0L);
- replay(memtableBytesFlushedSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(1L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(4L);
- memtableHitRatioSensor.record((double) 4 / (4 + 6), 0L);
- replay(memtableHitRatioSensor);
+ final long now = 0L;
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(1L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(2L);
+ doNothing().when(bytesWrittenToDatabaseSensor).record(1 + 2, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(3L);
+ doNothing().when(bytesReadFromDatabaseSensor).record(2 + 3, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(memtableBytesFlushedSensor).record(3 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(1L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(4L);
+ doNothing().when(memtableHitRatioSensor).record((double) 4 / (4 + 6), now);
final HistogramData memtableFlushTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 10L, 3.0);
final HistogramData memtableFlushTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 20.0, 4L, 8L, 10.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData2);
- memtableAvgFlushTimeSensor.record((double) (10 + 8) / (2 + 4), 0L);
- replay(memtableAvgFlushTimeSensor);
- memtableMinFlushTimeSensor.record(3.0, 0L);
- replay(memtableMinFlushTimeSensor);
- memtableMaxFlushTimeSensor.record(20.0, 0L);
- replay(memtableMaxFlushTimeSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(5L);
- writeStallDurationSensor.record(4 + 5, 0L);
- replay(writeStallDurationSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(2L);
- blockCacheDataHitRatioSensor.record((double) 8 / (8 + 6), 0L);
- replay(blockCacheDataHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(4L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(4L);
- blockCacheIndexHitRatioSensor.record((double) 6 / (6 + 6), 0L);
- replay(blockCacheIndexHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(2L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(5L);
- blockCacheFilterHitRatioSensor.record((double) 5 / (5 + 9), 0L);
- replay(blockCacheFilterHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(4L);
- bytesWrittenDuringCompactionSensor.record(2 + 4, 0L);
- replay(bytesWrittenDuringCompactionSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(5L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(6L);
- bytesReadDuringCompactionSensor.record(5 + 6, 0L);
- replay(bytesReadDuringCompactionSensor);
+ when(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData2);
+ doNothing().when(memtableAvgFlushTimeSensor).record((double) (10 + 8) / (2 + 4), now);
+ doNothing().when(memtableMinFlushTimeSensor).record(3.0, now);
+ doNothing().when(memtableMaxFlushTimeSensor).record(20.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(5L);
+ doNothing().when(writeStallDurationSensor).record(4 + 5, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(2L);
+ doNothing().when(blockCacheDataHitRatioSensor).record((double) 8 / (8 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(4L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(4L);
+ doNothing().when(blockCacheIndexHitRatioSensor).record((double) 6 / (6 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(2L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(5L);
+ doNothing().when(blockCacheFilterHitRatioSensor).record((double) 5 / (5 + 9), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(bytesWrittenDuringCompactionSensor).record(2 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(5L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(6L);
+ doNothing().when(bytesReadDuringCompactionSensor).record(5 + 6, now);
final HistogramData compactionTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 8L, 6.0);
final HistogramData compactionTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 24.0, 2L, 8L, 4.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData2);
- compactionTimeAvgSensor.record((double) (8 + 8) / (2 + 2), 0L);
- replay(compactionTimeAvgSensor);
- compactionTimeMinSensor.record(4.0, 0L);
- replay(compactionTimeMinSensor);
- compactionTimeMaxSensor.record(24.0, 0L);
- replay(compactionTimeMaxSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(7L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(4L);
- numberOfOpenFilesSensor.record((5 + 7) - (3 + 4), 0L);
- replay(numberOfOpenFilesSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(34L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(11L);
- numberOfFileErrorsSensor.record(11 + 34, 0L);
- replay(numberOfFileErrorsSensor);
-
- replay(statisticsToAdd1);
- replay(statisticsToAdd2);
-
- recorder.record(0L);
-
- verify(statisticsToAdd1);
- verify(statisticsToAdd2);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ when(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData2);
+ doNothing().when(compactionTimeAvgSensor).record((double) (8 + 8) / (2 + 2), now);
+ doNothing().when(compactionTimeMinSensor).record(4.0, now);
+ doNothing().when(compactionTimeMaxSensor).record(24.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(7L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(4L);
+ doNothing().when(numberOfOpenFilesSensor).record((5 + 7) - (3 + 4), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(34L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(11L);
+ doNothing().when(numberOfFileErrorsSensor).record(11 + 34, now);
+
+ recorder.record(now);
+
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(bytesWrittenToDatabaseSensor).record(isA(Double.class), eq(now));
+ verify(bytesReadFromDatabaseSensor).record(isA(Double.class), eq(now));
+ verify(memtableBytesFlushedSensor).record(isA(Double.class), eq(now));
+ verify(memtableHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(memtableAvgFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(memtableMinFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(memtableMaxFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(writeStallDurationSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheDataHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheIndexHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheFilterHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(bytesWrittenDuringCompactionSensor).record(isA(Double.class), eq(now));
+ verify(bytesReadDuringCompactionSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeAvgSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeMinSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeMaxSensor).record(isA(Double.class), eq(now));
+ verify(numberOfOpenFilesSensor).record(isA(Double.class), eq(now));
+ verify(numberOfFileErrorsSensor).record(isA(Double.class), eq(now));
}
@Test
public void shouldNotRecordStatisticsBasedMetricsIfStatisticsIsNull() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- replay(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
recorder.record(0L);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ verifyNoInteractions(bytesWrittenToDatabaseSensor,
+ bytesReadFromDatabaseSensor,
+ memtableBytesFlushedSensor,
+ memtableHitRatioSensor,
+ memtableAvgFlushTimeSensor,
+ memtableMinFlushTimeSensor,
+ memtableMaxFlushTimeSensor,
+ writeStallDurationSensor,
+ blockCacheDataHitRatioSensor,
+ blockCacheIndexHitRatioSensor,
+ blockCacheFilterHitRatioSensor,
+ bytesWrittenDuringCompactionSensor,
+ bytesReadDuringCompactionSensor,
+ compactionTimeAvgSensor,
+ compactionTimeMinSensor,
+ compactionTimeMaxSensor,
+ numberOfOpenFilesSensor,
+ numberOfFileErrorsSensor);
}
@Test
public void shouldCorrectlyHandleHitRatioRecordingsWithZeroHitsAndMisses() {
- reset(statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- expect(statisticsToAdd1.getHistogramData(anyObject())).andStubReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
- expect(statisticsToAdd1.getAndResetTickerCount(anyObject())).andStubReturn(0L);
- replay(statisticsToAdd1);
- memtableHitRatioSensor.record(0, 0L);
- blockCacheDataHitRatioSensor.record(0, 0L);
- blockCacheIndexHitRatioSensor.record(0, 0L);
- blockCacheFilterHitRatioSensor.record(0, 0L);
- replay(memtableHitRatioSensor);
- replay(blockCacheDataHitRatioSensor);
- replay(blockCacheIndexHitRatioSensor);
- replay(blockCacheFilterHitRatioSensor);
+ when(statisticsToAdd1.getHistogramData(any())).thenReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
+ when(statisticsToAdd1.getAndResetTickerCount(any())).thenReturn(0L);
recorder.record(0L);
- verify(memtableHitRatioSensor);
- verify(blockCacheDataHitRatioSensor);
- verify(blockCacheIndexHitRatioSensor);
- verify(blockCacheFilterHitRatioSensor);
+ verify(memtableHitRatioSensor).record(0d, 0L);
+ verify(blockCacheDataHitRatioSensor).record(0d, 0L);
+ verify(blockCacheIndexHitRatioSensor).record(0d, 0L);
+ verify(blockCacheFilterHitRatioSensor).record(0d, 0L);
+ verify(statisticsToAdd1, times(2)).getHistogramData(any());
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(any());
Review Comment:
Those verifications are not needed since you have already done them in `shouldRecordStatisticsBasedMetrics()`. This method tests the computation of the hit ratio and not the retrieval of the single values.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -70,105 +73,133 @@ public class RocksDBMetricsRecorderTest {
private final Statistics statisticsToAdd2 = mock(Statistics.class);
private final Statistics statisticsToAdd3 = mock(Statistics.class);
- private final Sensor bytesWrittenToDatabaseSensor = createMock(Sensor.class);
- private final Sensor bytesReadFromDatabaseSensor = createMock(Sensor.class);
- private final Sensor memtableBytesFlushedSensor = createMock(Sensor.class);
- private final Sensor memtableHitRatioSensor = createMock(Sensor.class);
- private final Sensor memtableAvgFlushTimeSensor = createMock(Sensor.class);
- private final Sensor memtableMinFlushTimeSensor = createMock(Sensor.class);
- private final Sensor memtableMaxFlushTimeSensor = createMock(Sensor.class);
- private final Sensor writeStallDurationSensor = createMock(Sensor.class);
- private final Sensor blockCacheDataHitRatioSensor = createMock(Sensor.class);
- private final Sensor blockCacheIndexHitRatioSensor = createMock(Sensor.class);
- private final Sensor blockCacheFilterHitRatioSensor = createMock(Sensor.class);
- private final Sensor bytesReadDuringCompactionSensor = createMock(Sensor.class);
- private final Sensor bytesWrittenDuringCompactionSensor = createMock(Sensor.class);
- private final Sensor compactionTimeAvgSensor = createMock(Sensor.class);
- private final Sensor compactionTimeMinSensor = createMock(Sensor.class);
- private final Sensor compactionTimeMaxSensor = createMock(Sensor.class);
- private final Sensor numberOfOpenFilesSensor = createMock(Sensor.class);
- private final Sensor numberOfFileErrorsSensor = createMock(Sensor.class);
-
- private final StreamsMetricsImpl streamsMetrics = niceMock(StreamsMetricsImpl.class);
+ private final Sensor bytesWrittenToDatabaseSensor = mock(Sensor.class);
+ private final Sensor bytesReadFromDatabaseSensor = mock(Sensor.class);
+ private final Sensor memtableBytesFlushedSensor = mock(Sensor.class);
+ private final Sensor memtableHitRatioSensor = mock(Sensor.class);
+ private final Sensor memtableAvgFlushTimeSensor = mock(Sensor.class);
+ private final Sensor memtableMinFlushTimeSensor = mock(Sensor.class);
+ private final Sensor memtableMaxFlushTimeSensor = mock(Sensor.class);
+ private final Sensor writeStallDurationSensor = mock(Sensor.class);
+ private final Sensor blockCacheDataHitRatioSensor = mock(Sensor.class);
+ private final Sensor blockCacheIndexHitRatioSensor = mock(Sensor.class);
+ private final Sensor blockCacheFilterHitRatioSensor = mock(Sensor.class);
+ private final Sensor bytesReadDuringCompactionSensor = mock(Sensor.class);
+ private final Sensor bytesWrittenDuringCompactionSensor = mock(Sensor.class);
+ private final Sensor compactionTimeAvgSensor = mock(Sensor.class);
+ private final Sensor compactionTimeMinSensor = mock(Sensor.class);
+ private final Sensor compactionTimeMaxSensor = mock(Sensor.class);
+ private final Sensor numberOfOpenFilesSensor = mock(Sensor.class);
+ private final Sensor numberOfFileErrorsSensor = mock(Sensor.class);
+
+ private final StreamsMetricsImpl streamsMetrics = mock(StreamsMetricsImpl.class);
private final RocksDBMetricsRecordingTrigger recordingTrigger = mock(RocksDBMetricsRecordingTrigger.class);
private final RocksDBMetricsRecorder recorder = new RocksDBMetricsRecorder(METRICS_SCOPE, STORE_NAME);
+ private MockedStatic<RocksDBMetrics> dbMetrics;
+
@Before
public void setUp() {
- setUpMetricsStubMock();
- expect(streamsMetrics.rocksDBMetricsRecordingTrigger()).andStubReturn(recordingTrigger);
- replay(streamsMetrics);
+ setUpMetricsMock();
+ when(streamsMetrics.rocksDBMetricsRecordingTrigger()).thenReturn(recordingTrigger);
recorder.init(streamsMetrics, TASK_ID1);
}
+ @After
+ public void cleanUpMocks() {
+ dbMetrics.close();
Review Comment:
The Mockito docs say:
> To make sure a static mock remains temporary, it is recommended to define the scope within a try-with-resources construct.
https://javadoc.io/doc/org.mockito/mockito-core/latest/org/mockito/Mockito.html#static_mocks
So I think it would make the code more robust if we used a try-with-resource construct.
Test method `shouldSetStatsLevelToExceptDetailedTimersWhenValueProvidersWithStatisticsAreAdded()` would become
```
@Test
public void shouldSetStatsLevelToExceptDetailedTimersWhenValueProvidersWithStatisticsAreAdded() {
final RocksDBMetricContext metricsContext =
new RocksDBMetricContext(TASK_ID1.toString(), METRICS_SCOPE, STORE_NAME);
try (final MockedStatic<RocksDBMetrics> dbMetrics = mockStatic(RocksDBMetrics.class)) {
setUpMetricsMock(dbMetrics, metricsContext);
recorder.init(streamsMetrics, TASK_ID1);
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
verify(statisticsToAdd1).setStatsLevel(StatsLevel.EXCEPT_DETAILED_TIMERS);
}
}
```
where `metricsContext` might probably be specified as a test object field depedent on the other test methods. Maybe you can even move the whole try-with-resource construct into the `setUp()` method.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
Review Comment:
Could you please remove the trailing `2` from the test name? I do not know why it was added.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- verify(recordingTrigger);
+ verify(recordingTrigger, never()).removeMetricsRecorder(recorder);
- reset(recordingTrigger);
- recordingTrigger.removeMetricsRecorder(recorder);
- replay(recordingTrigger);
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_2);
- verify(recordingTrigger);
+ verify(recordingTrigger).removeMetricsRecorder(recorder);
}
@Test
public void shouldThrowIfValueProvidersToRemoveNotFound() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
assertThrows(
- IllegalStateException.class,
- () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
+ IllegalStateException.class,
+ () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
);
}
@Test
public void shouldRecordStatisticsBasedMetrics() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(statisticsToAdd1);
- reset(statisticsToAdd2);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(1L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(2L);
- bytesWrittenToDatabaseSensor.record(1 + 2, 0L);
- replay(bytesWrittenToDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(3L);
- bytesReadFromDatabaseSensor.record(2 + 3, 0L);
- replay(bytesReadFromDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(4L);
- memtableBytesFlushedSensor.record(3 + 4, 0L);
- replay(memtableBytesFlushedSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(1L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(4L);
- memtableHitRatioSensor.record((double) 4 / (4 + 6), 0L);
- replay(memtableHitRatioSensor);
+ final long now = 0L;
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(1L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(2L);
+ doNothing().when(bytesWrittenToDatabaseSensor).record(1 + 2, now);
Review Comment:
I do not think this is needed. As it says, it does nothing 🙂. It also does not verify anything. Same applies to the other stubbings of void methods.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- verify(recordingTrigger);
+ verify(recordingTrigger, never()).removeMetricsRecorder(recorder);
- reset(recordingTrigger);
- recordingTrigger.removeMetricsRecorder(recorder);
- replay(recordingTrigger);
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_2);
- verify(recordingTrigger);
+ verify(recordingTrigger).removeMetricsRecorder(recorder);
}
@Test
public void shouldThrowIfValueProvidersToRemoveNotFound() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
assertThrows(
- IllegalStateException.class,
- () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
+ IllegalStateException.class,
+ () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
);
}
@Test
public void shouldRecordStatisticsBasedMetrics() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(statisticsToAdd1);
- reset(statisticsToAdd2);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(1L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(2L);
- bytesWrittenToDatabaseSensor.record(1 + 2, 0L);
- replay(bytesWrittenToDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(3L);
- bytesReadFromDatabaseSensor.record(2 + 3, 0L);
- replay(bytesReadFromDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(4L);
- memtableBytesFlushedSensor.record(3 + 4, 0L);
- replay(memtableBytesFlushedSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(1L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(4L);
- memtableHitRatioSensor.record((double) 4 / (4 + 6), 0L);
- replay(memtableHitRatioSensor);
+ final long now = 0L;
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(1L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(2L);
+ doNothing().when(bytesWrittenToDatabaseSensor).record(1 + 2, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(3L);
+ doNothing().when(bytesReadFromDatabaseSensor).record(2 + 3, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(memtableBytesFlushedSensor).record(3 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(1L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(4L);
+ doNothing().when(memtableHitRatioSensor).record((double) 4 / (4 + 6), now);
final HistogramData memtableFlushTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 10L, 3.0);
final HistogramData memtableFlushTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 20.0, 4L, 8L, 10.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData2);
- memtableAvgFlushTimeSensor.record((double) (10 + 8) / (2 + 4), 0L);
- replay(memtableAvgFlushTimeSensor);
- memtableMinFlushTimeSensor.record(3.0, 0L);
- replay(memtableMinFlushTimeSensor);
- memtableMaxFlushTimeSensor.record(20.0, 0L);
- replay(memtableMaxFlushTimeSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(5L);
- writeStallDurationSensor.record(4 + 5, 0L);
- replay(writeStallDurationSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(2L);
- blockCacheDataHitRatioSensor.record((double) 8 / (8 + 6), 0L);
- replay(blockCacheDataHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(4L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(4L);
- blockCacheIndexHitRatioSensor.record((double) 6 / (6 + 6), 0L);
- replay(blockCacheIndexHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(2L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(5L);
- blockCacheFilterHitRatioSensor.record((double) 5 / (5 + 9), 0L);
- replay(blockCacheFilterHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(4L);
- bytesWrittenDuringCompactionSensor.record(2 + 4, 0L);
- replay(bytesWrittenDuringCompactionSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(5L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(6L);
- bytesReadDuringCompactionSensor.record(5 + 6, 0L);
- replay(bytesReadDuringCompactionSensor);
+ when(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData2);
+ doNothing().when(memtableAvgFlushTimeSensor).record((double) (10 + 8) / (2 + 4), now);
+ doNothing().when(memtableMinFlushTimeSensor).record(3.0, now);
+ doNothing().when(memtableMaxFlushTimeSensor).record(20.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(5L);
+ doNothing().when(writeStallDurationSensor).record(4 + 5, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(2L);
+ doNothing().when(blockCacheDataHitRatioSensor).record((double) 8 / (8 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(4L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(4L);
+ doNothing().when(blockCacheIndexHitRatioSensor).record((double) 6 / (6 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(2L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(5L);
+ doNothing().when(blockCacheFilterHitRatioSensor).record((double) 5 / (5 + 9), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(bytesWrittenDuringCompactionSensor).record(2 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(5L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(6L);
+ doNothing().when(bytesReadDuringCompactionSensor).record(5 + 6, now);
final HistogramData compactionTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 8L, 6.0);
final HistogramData compactionTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 24.0, 2L, 8L, 4.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData2);
- compactionTimeAvgSensor.record((double) (8 + 8) / (2 + 2), 0L);
- replay(compactionTimeAvgSensor);
- compactionTimeMinSensor.record(4.0, 0L);
- replay(compactionTimeMinSensor);
- compactionTimeMaxSensor.record(24.0, 0L);
- replay(compactionTimeMaxSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(7L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(4L);
- numberOfOpenFilesSensor.record((5 + 7) - (3 + 4), 0L);
- replay(numberOfOpenFilesSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(34L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(11L);
- numberOfFileErrorsSensor.record(11 + 34, 0L);
- replay(numberOfFileErrorsSensor);
-
- replay(statisticsToAdd1);
- replay(statisticsToAdd2);
-
- recorder.record(0L);
-
- verify(statisticsToAdd1);
- verify(statisticsToAdd2);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ when(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData2);
+ doNothing().when(compactionTimeAvgSensor).record((double) (8 + 8) / (2 + 2), now);
+ doNothing().when(compactionTimeMinSensor).record(4.0, now);
+ doNothing().when(compactionTimeMaxSensor).record(24.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(7L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(4L);
+ doNothing().when(numberOfOpenFilesSensor).record((5 + 7) - (3 + 4), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(34L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(11L);
+ doNothing().when(numberOfFileErrorsSensor).record(11 + 34, now);
+
+ recorder.record(now);
+
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(bytesWrittenToDatabaseSensor).record(isA(Double.class), eq(now));
+ verify(bytesReadFromDatabaseSensor).record(isA(Double.class), eq(now));
+ verify(memtableBytesFlushedSensor).record(isA(Double.class), eq(now));
+ verify(memtableHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(memtableAvgFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(memtableMinFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(memtableMaxFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(writeStallDurationSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheDataHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheIndexHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheFilterHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(bytesWrittenDuringCompactionSensor).record(isA(Double.class), eq(now));
+ verify(bytesReadDuringCompactionSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeAvgSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeMinSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeMaxSensor).record(isA(Double.class), eq(now));
+ verify(numberOfOpenFilesSensor).record(isA(Double.class), eq(now));
+ verify(numberOfFileErrorsSensor).record(isA(Double.class), eq(now));
}
@Test
public void shouldNotRecordStatisticsBasedMetricsIfStatisticsIsNull() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- replay(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
recorder.record(0L);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ verifyNoInteractions(bytesWrittenToDatabaseSensor,
+ bytesReadFromDatabaseSensor,
+ memtableBytesFlushedSensor,
+ memtableHitRatioSensor,
+ memtableAvgFlushTimeSensor,
+ memtableMinFlushTimeSensor,
+ memtableMaxFlushTimeSensor,
+ writeStallDurationSensor,
+ blockCacheDataHitRatioSensor,
+ blockCacheIndexHitRatioSensor,
+ blockCacheFilterHitRatioSensor,
+ bytesWrittenDuringCompactionSensor,
+ bytesReadDuringCompactionSensor,
+ compactionTimeAvgSensor,
+ compactionTimeMinSensor,
+ compactionTimeMaxSensor,
+ numberOfOpenFilesSensor,
+ numberOfFileErrorsSensor);
}
@Test
public void shouldCorrectlyHandleHitRatioRecordingsWithZeroHitsAndMisses() {
- reset(statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- expect(statisticsToAdd1.getHistogramData(anyObject())).andStubReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
- expect(statisticsToAdd1.getAndResetTickerCount(anyObject())).andStubReturn(0L);
- replay(statisticsToAdd1);
- memtableHitRatioSensor.record(0, 0L);
- blockCacheDataHitRatioSensor.record(0, 0L);
- blockCacheIndexHitRatioSensor.record(0, 0L);
- blockCacheFilterHitRatioSensor.record(0, 0L);
- replay(memtableHitRatioSensor);
- replay(blockCacheDataHitRatioSensor);
- replay(blockCacheIndexHitRatioSensor);
- replay(blockCacheFilterHitRatioSensor);
+ when(statisticsToAdd1.getHistogramData(any())).thenReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
+ when(statisticsToAdd1.getAndResetTickerCount(any())).thenReturn(0L);
recorder.record(0L);
- verify(memtableHitRatioSensor);
- verify(blockCacheDataHitRatioSensor);
- verify(blockCacheIndexHitRatioSensor);
- verify(blockCacheFilterHitRatioSensor);
+ verify(memtableHitRatioSensor).record(0d, 0L);
+ verify(blockCacheDataHitRatioSensor).record(0d, 0L);
+ verify(blockCacheIndexHitRatioSensor).record(0d, 0L);
+ verify(blockCacheFilterHitRatioSensor).record(0d, 0L);
+ verify(statisticsToAdd1, times(2)).getHistogramData(any());
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(any());
}
@Test
public void shouldCorrectlyHandleAvgRecordingsWithZeroSumAndCount() {
- reset(statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- expect(statisticsToAdd1.getHistogramData(anyObject())).andStubReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
- expect(statisticsToAdd1.getAndResetTickerCount(anyObject())).andStubReturn(0L);
- replay(statisticsToAdd1);
- memtableAvgFlushTimeSensor.record(0, 0L);
- compactionTimeAvgSensor.record(0, 0L);
- replay(memtableAvgFlushTimeSensor);
- replay(compactionTimeAvgSensor);
+ final long now = 0L;
+ when(statisticsToAdd1.getHistogramData(any())).thenReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
+ when(statisticsToAdd1.getAndResetTickerCount(any())).thenReturn(0L);
- recorder.record(0L);
+ recorder.record(now);
- verify(memtableAvgFlushTimeSensor);
- verify(compactionTimeAvgSensor);
+ verify(compactionTimeAvgSensor).record(0d, now);
+ verify(memtableAvgFlushTimeSensor).record(0d, now);
+ verify(statisticsToAdd1, times(2)).getHistogramData(any());
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(any());
}
private void setUpMetricsMock() {
- mockStatic(RocksDBMetrics.class);
- final RocksDBMetricContext metricsContext =
- new RocksDBMetricContext(TASK_ID1.toString(), METRICS_SCOPE, STORE_NAME);
- expect(RocksDBMetrics.bytesWrittenToDatabaseSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(bytesWrittenToDatabaseSensor);
- expect(RocksDBMetrics.bytesReadFromDatabaseSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(bytesReadFromDatabaseSensor);
- expect(RocksDBMetrics.memtableBytesFlushedSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(memtableBytesFlushedSensor);
- expect(RocksDBMetrics.memtableHitRatioSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(memtableHitRatioSensor);
- expect(RocksDBMetrics.memtableAvgFlushTimeSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(memtableAvgFlushTimeSensor);
- expect(RocksDBMetrics.memtableMinFlushTimeSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(memtableMinFlushTimeSensor);
- expect(RocksDBMetrics.memtableMaxFlushTimeSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(memtableMaxFlushTimeSensor);
- expect(RocksDBMetrics.writeStallDurationSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(writeStallDurationSensor);
- expect(RocksDBMetrics.blockCacheDataHitRatioSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(blockCacheDataHitRatioSensor);
- expect(RocksDBMetrics.blockCacheIndexHitRatioSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(blockCacheIndexHitRatioSensor);
- expect(RocksDBMetrics.blockCacheFilterHitRatioSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(blockCacheFilterHitRatioSensor);
- expect(RocksDBMetrics.bytesWrittenDuringCompactionSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(bytesWrittenDuringCompactionSensor);
- expect(RocksDBMetrics.bytesReadDuringCompactionSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(bytesReadDuringCompactionSensor);
- expect(RocksDBMetrics.compactionTimeAvgSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(compactionTimeAvgSensor);
- expect(RocksDBMetrics.compactionTimeMinSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(compactionTimeMinSensor);
- expect(RocksDBMetrics.compactionTimeMaxSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(compactionTimeMaxSensor);
- expect(RocksDBMetrics.numberOfOpenFilesSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(numberOfOpenFilesSensor);
- expect(RocksDBMetrics.numberOfFileErrorsSensor(eq(streamsMetrics), eq(metricsContext)))
- .andReturn(numberOfFileErrorsSensor);
- RocksDBMetrics.addNumImmutableMemTableMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addCurSizeActiveMemTable(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addCurSizeAllMemTables(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addSizeAllMemTables(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumEntriesActiveMemTableMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumEntriesImmMemTablesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumDeletesActiveMemTableMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumDeletesImmMemTablesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addMemTableFlushPending(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumRunningFlushesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addCompactionPendingMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumRunningCompactionsMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addEstimatePendingCompactionBytesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addTotalSstFilesSizeMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addLiveSstFilesSizeMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumLiveVersionMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBlockCacheCapacityMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBlockCacheUsageMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBlockCachePinnedUsageMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addEstimateNumKeysMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addEstimateTableReadersMemMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBackgroundErrorsMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- replay(RocksDBMetrics.class);
- }
-
- private void setUpMetricsStubMock() {
- mockStatic(RocksDBMetrics.class);
+ dbMetrics = mockStatic(RocksDBMetrics.class);
final RocksDBMetricContext metricsContext =
- new RocksDBMetricContext(TASK_ID1.toString(), METRICS_SCOPE, STORE_NAME);
- expect(RocksDBMetrics.bytesWrittenToDatabaseSensor(streamsMetrics, metricsContext))
- .andStubReturn(bytesWrittenToDatabaseSensor);
- expect(RocksDBMetrics.bytesReadFromDatabaseSensor(streamsMetrics, metricsContext))
- .andStubReturn(bytesReadFromDatabaseSensor);
- expect(RocksDBMetrics.memtableBytesFlushedSensor(streamsMetrics, metricsContext))
- .andStubReturn(memtableBytesFlushedSensor);
- expect(RocksDBMetrics.memtableHitRatioSensor(streamsMetrics, metricsContext))
- .andStubReturn(memtableHitRatioSensor);
- expect(RocksDBMetrics.memtableAvgFlushTimeSensor(streamsMetrics, metricsContext))
- .andStubReturn(memtableAvgFlushTimeSensor);
- expect(RocksDBMetrics.memtableMinFlushTimeSensor(streamsMetrics, metricsContext))
- .andStubReturn(memtableMinFlushTimeSensor);
- expect(RocksDBMetrics.memtableMaxFlushTimeSensor(streamsMetrics, metricsContext))
- .andStubReturn(memtableMaxFlushTimeSensor);
- expect(RocksDBMetrics.writeStallDurationSensor(streamsMetrics, metricsContext))
- .andStubReturn(writeStallDurationSensor);
- expect(RocksDBMetrics.blockCacheDataHitRatioSensor(streamsMetrics, metricsContext))
- .andStubReturn(blockCacheDataHitRatioSensor);
- expect(RocksDBMetrics.blockCacheIndexHitRatioSensor(streamsMetrics, metricsContext))
- .andStubReturn(blockCacheIndexHitRatioSensor);
- expect(RocksDBMetrics.blockCacheFilterHitRatioSensor(streamsMetrics, metricsContext))
- .andStubReturn(blockCacheFilterHitRatioSensor);
- expect(RocksDBMetrics.bytesWrittenDuringCompactionSensor(streamsMetrics, metricsContext))
- .andStubReturn(bytesWrittenDuringCompactionSensor);
- expect(RocksDBMetrics.bytesReadDuringCompactionSensor(streamsMetrics, metricsContext))
- .andStubReturn(bytesReadDuringCompactionSensor);
- expect(RocksDBMetrics.compactionTimeAvgSensor(streamsMetrics, metricsContext))
- .andStubReturn(compactionTimeAvgSensor);
- expect(RocksDBMetrics.compactionTimeMinSensor(streamsMetrics, metricsContext))
- .andStubReturn(compactionTimeMinSensor);
- expect(RocksDBMetrics.compactionTimeMaxSensor(streamsMetrics, metricsContext))
- .andStubReturn(compactionTimeMaxSensor);
- expect(RocksDBMetrics.numberOfOpenFilesSensor(streamsMetrics, metricsContext))
- .andStubReturn(numberOfOpenFilesSensor);
- expect(RocksDBMetrics.numberOfFileErrorsSensor(streamsMetrics, metricsContext))
- .andStubReturn(numberOfFileErrorsSensor);
- RocksDBMetrics.addNumImmutableMemTableMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addCurSizeActiveMemTable(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addCurSizeAllMemTables(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addSizeAllMemTables(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumEntriesActiveMemTableMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumEntriesImmMemTablesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumDeletesActiveMemTableMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumDeletesImmMemTablesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addMemTableFlushPending(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumRunningFlushesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addCompactionPendingMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumRunningCompactionsMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addEstimatePendingCompactionBytesMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addTotalSstFilesSizeMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addLiveSstFilesSizeMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addNumLiveVersionMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBlockCacheCapacityMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBlockCacheUsageMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBlockCachePinnedUsageMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addEstimateNumKeysMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addEstimateTableReadersMemMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- RocksDBMetrics.addBackgroundErrorsMetric(eq(streamsMetrics), eq(metricsContext), anyObject());
- replay(RocksDBMetrics.class);
+ new RocksDBMetricContext(TASK_ID1.toString(), METRICS_SCOPE, STORE_NAME);
+ dbMetrics.when(() -> RocksDBMetrics.bytesWrittenToDatabaseSensor(streamsMetrics, metricsContext))
+ .thenReturn(bytesWrittenToDatabaseSensor);
+ dbMetrics.when(() -> RocksDBMetrics.bytesReadFromDatabaseSensor(streamsMetrics, metricsContext))
+ .thenReturn(bytesReadFromDatabaseSensor);
+ dbMetrics.when(() -> RocksDBMetrics.memtableBytesFlushedSensor(streamsMetrics, metricsContext))
+ .thenReturn(memtableBytesFlushedSensor);
+ dbMetrics.when(() -> RocksDBMetrics.memtableHitRatioSensor(streamsMetrics, metricsContext))
+ .thenReturn(memtableHitRatioSensor);
+ dbMetrics.when(() -> RocksDBMetrics.memtableAvgFlushTimeSensor(streamsMetrics, metricsContext))
+ .thenReturn(memtableAvgFlushTimeSensor);
+ dbMetrics.when(() -> RocksDBMetrics.memtableMinFlushTimeSensor(streamsMetrics, metricsContext))
+ .thenReturn(memtableMinFlushTimeSensor);
+ dbMetrics.when(() -> RocksDBMetrics.memtableMaxFlushTimeSensor(streamsMetrics, metricsContext))
+ .thenReturn(memtableMaxFlushTimeSensor);
+ dbMetrics.when(() -> RocksDBMetrics.writeStallDurationSensor(streamsMetrics, metricsContext))
+ .thenReturn(writeStallDurationSensor);
+ dbMetrics.when(() -> RocksDBMetrics.blockCacheDataHitRatioSensor(streamsMetrics, metricsContext))
+ .thenReturn(blockCacheDataHitRatioSensor);
+ dbMetrics.when(() -> RocksDBMetrics.blockCacheIndexHitRatioSensor(streamsMetrics, metricsContext))
+ .thenReturn(blockCacheIndexHitRatioSensor);
+ dbMetrics.when(() -> RocksDBMetrics.blockCacheFilterHitRatioSensor(streamsMetrics, metricsContext))
+ .thenReturn(blockCacheFilterHitRatioSensor);
+ dbMetrics.when(() -> RocksDBMetrics.bytesWrittenDuringCompactionSensor(streamsMetrics, metricsContext))
+ .thenReturn(bytesWrittenDuringCompactionSensor);
+ dbMetrics.when(() -> RocksDBMetrics.bytesReadDuringCompactionSensor(streamsMetrics, metricsContext))
+ .thenReturn(bytesReadDuringCompactionSensor);
+ dbMetrics.when(() -> RocksDBMetrics.compactionTimeAvgSensor(streamsMetrics, metricsContext))
+ .thenReturn(compactionTimeAvgSensor);
+ dbMetrics.when(() -> RocksDBMetrics.compactionTimeMinSensor(streamsMetrics, metricsContext))
+ .thenReturn(compactionTimeMinSensor);
+ dbMetrics.when(() -> RocksDBMetrics.compactionTimeMaxSensor(streamsMetrics, metricsContext))
+ .thenReturn(compactionTimeMaxSensor);
+ dbMetrics.when(() -> RocksDBMetrics.numberOfOpenFilesSensor(streamsMetrics, metricsContext))
+ .thenReturn(numberOfOpenFilesSensor);
+ dbMetrics.when(() -> RocksDBMetrics.numberOfFileErrorsSensor(streamsMetrics, metricsContext))
+ .thenReturn(numberOfFileErrorsSensor);
+ dbMetrics.when(() -> RocksDBMetrics.addNumImmutableMemTableMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addCurSizeActiveMemTable(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addCurSizeAllMemTables(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addSizeAllMemTables(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumEntriesActiveMemTableMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumEntriesImmMemTablesMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumDeletesActiveMemTableMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumDeletesImmMemTablesMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addMemTableFlushPending(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumRunningFlushesMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addCompactionPendingMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumRunningCompactionsMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addEstimatePendingCompactionBytesMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addTotalSstFilesSizeMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addLiveSstFilesSizeMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addNumLiveVersionMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addBlockCacheCapacityMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addBlockCacheUsageMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addBlockCachePinnedUsageMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addEstimateNumKeysMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addEstimateTableReadersMemMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
+ dbMetrics.when(() -> RocksDBMetrics.addBackgroundErrorsMetric(eq(streamsMetrics), eq(metricsContext), any()))
+ .thenAnswer(invocation -> null);
Review Comment:
I think
```
dbMetrics.when(() -> RocksDBMetrics.addNumImmutableMemTableMetric(eq(streamsMetrics), eq(metricsContext), any())).thenAnswer(invocation -> null);
...
```
```
dbMetrics.verify(() -> RocksDBMetrics.addNumImmutableMemTableMetric(eq(streamsMetrics), eq(metricsContext), any()));
...
```
We just want to verify that these methods are called during `recorder.init()`. So you can extract this verifications to a method and only call this method in `shouldInitMetricsRecorder()`. With EasyMock these calls needed to be here since EasyMock does not allow to not verify/specify an actual call in default mode.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- verify(recordingTrigger);
+ verify(recordingTrigger, never()).removeMetricsRecorder(recorder);
- reset(recordingTrigger);
- recordingTrigger.removeMetricsRecorder(recorder);
- replay(recordingTrigger);
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_2);
- verify(recordingTrigger);
+ verify(recordingTrigger).removeMetricsRecorder(recorder);
}
@Test
public void shouldThrowIfValueProvidersToRemoveNotFound() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
assertThrows(
- IllegalStateException.class,
- () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
+ IllegalStateException.class,
+ () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
);
}
@Test
public void shouldRecordStatisticsBasedMetrics() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(statisticsToAdd1);
- reset(statisticsToAdd2);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(1L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(2L);
- bytesWrittenToDatabaseSensor.record(1 + 2, 0L);
- replay(bytesWrittenToDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(3L);
- bytesReadFromDatabaseSensor.record(2 + 3, 0L);
- replay(bytesReadFromDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(4L);
- memtableBytesFlushedSensor.record(3 + 4, 0L);
- replay(memtableBytesFlushedSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(1L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(4L);
- memtableHitRatioSensor.record((double) 4 / (4 + 6), 0L);
- replay(memtableHitRatioSensor);
+ final long now = 0L;
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(1L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(2L);
+ doNothing().when(bytesWrittenToDatabaseSensor).record(1 + 2, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(3L);
+ doNothing().when(bytesReadFromDatabaseSensor).record(2 + 3, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(memtableBytesFlushedSensor).record(3 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(1L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(4L);
+ doNothing().when(memtableHitRatioSensor).record((double) 4 / (4 + 6), now);
final HistogramData memtableFlushTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 10L, 3.0);
final HistogramData memtableFlushTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 20.0, 4L, 8L, 10.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData2);
- memtableAvgFlushTimeSensor.record((double) (10 + 8) / (2 + 4), 0L);
- replay(memtableAvgFlushTimeSensor);
- memtableMinFlushTimeSensor.record(3.0, 0L);
- replay(memtableMinFlushTimeSensor);
- memtableMaxFlushTimeSensor.record(20.0, 0L);
- replay(memtableMaxFlushTimeSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(5L);
- writeStallDurationSensor.record(4 + 5, 0L);
- replay(writeStallDurationSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(2L);
- blockCacheDataHitRatioSensor.record((double) 8 / (8 + 6), 0L);
- replay(blockCacheDataHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(4L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(4L);
- blockCacheIndexHitRatioSensor.record((double) 6 / (6 + 6), 0L);
- replay(blockCacheIndexHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(2L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(5L);
- blockCacheFilterHitRatioSensor.record((double) 5 / (5 + 9), 0L);
- replay(blockCacheFilterHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(4L);
- bytesWrittenDuringCompactionSensor.record(2 + 4, 0L);
- replay(bytesWrittenDuringCompactionSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(5L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(6L);
- bytesReadDuringCompactionSensor.record(5 + 6, 0L);
- replay(bytesReadDuringCompactionSensor);
+ when(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData2);
+ doNothing().when(memtableAvgFlushTimeSensor).record((double) (10 + 8) / (2 + 4), now);
+ doNothing().when(memtableMinFlushTimeSensor).record(3.0, now);
+ doNothing().when(memtableMaxFlushTimeSensor).record(20.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(5L);
+ doNothing().when(writeStallDurationSensor).record(4 + 5, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(2L);
+ doNothing().when(blockCacheDataHitRatioSensor).record((double) 8 / (8 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(4L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(4L);
+ doNothing().when(blockCacheIndexHitRatioSensor).record((double) 6 / (6 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(2L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(5L);
+ doNothing().when(blockCacheFilterHitRatioSensor).record((double) 5 / (5 + 9), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(bytesWrittenDuringCompactionSensor).record(2 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(5L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(6L);
+ doNothing().when(bytesReadDuringCompactionSensor).record(5 + 6, now);
final HistogramData compactionTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 8L, 6.0);
final HistogramData compactionTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 24.0, 2L, 8L, 4.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData2);
- compactionTimeAvgSensor.record((double) (8 + 8) / (2 + 2), 0L);
- replay(compactionTimeAvgSensor);
- compactionTimeMinSensor.record(4.0, 0L);
- replay(compactionTimeMinSensor);
- compactionTimeMaxSensor.record(24.0, 0L);
- replay(compactionTimeMaxSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(7L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(4L);
- numberOfOpenFilesSensor.record((5 + 7) - (3 + 4), 0L);
- replay(numberOfOpenFilesSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(34L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(11L);
- numberOfFileErrorsSensor.record(11 + 34, 0L);
- replay(numberOfFileErrorsSensor);
-
- replay(statisticsToAdd1);
- replay(statisticsToAdd2);
-
- recorder.record(0L);
-
- verify(statisticsToAdd1);
- verify(statisticsToAdd2);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ when(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData2);
+ doNothing().when(compactionTimeAvgSensor).record((double) (8 + 8) / (2 + 2), now);
+ doNothing().when(compactionTimeMinSensor).record(4.0, now);
+ doNothing().when(compactionTimeMaxSensor).record(24.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(7L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(4L);
+ doNothing().when(numberOfOpenFilesSensor).record((5 + 7) - (3 + 4), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(34L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(11L);
+ doNothing().when(numberOfFileErrorsSensor).record(11 + 34, now);
+
+ recorder.record(now);
+
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(bytesWrittenToDatabaseSensor).record(isA(Double.class), eq(now));
+ verify(bytesReadFromDatabaseSensor).record(isA(Double.class), eq(now));
+ verify(memtableBytesFlushedSensor).record(isA(Double.class), eq(now));
+ verify(memtableHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(memtableAvgFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(memtableMinFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(memtableMaxFlushTimeSensor).record(isA(Double.class), eq(now));
+ verify(writeStallDurationSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheDataHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheIndexHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(blockCacheFilterHitRatioSensor).record(isA(Double.class), eq(now));
+ verify(bytesWrittenDuringCompactionSensor).record(isA(Double.class), eq(now));
+ verify(bytesReadDuringCompactionSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeAvgSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeMinSensor).record(isA(Double.class), eq(now));
+ verify(compactionTimeMaxSensor).record(isA(Double.class), eq(now));
+ verify(numberOfOpenFilesSensor).record(isA(Double.class), eq(now));
+ verify(numberOfFileErrorsSensor).record(isA(Double.class), eq(now));
}
@Test
public void shouldNotRecordStatisticsBasedMetricsIfStatisticsIsNull() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- replay(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
recorder.record(0L);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ verifyNoInteractions(bytesWrittenToDatabaseSensor,
+ bytesReadFromDatabaseSensor,
+ memtableBytesFlushedSensor,
+ memtableHitRatioSensor,
+ memtableAvgFlushTimeSensor,
+ memtableMinFlushTimeSensor,
+ memtableMaxFlushTimeSensor,
+ writeStallDurationSensor,
+ blockCacheDataHitRatioSensor,
+ blockCacheIndexHitRatioSensor,
+ blockCacheFilterHitRatioSensor,
+ bytesWrittenDuringCompactionSensor,
+ bytesReadDuringCompactionSensor,
+ compactionTimeAvgSensor,
+ compactionTimeMinSensor,
+ compactionTimeMaxSensor,
+ numberOfOpenFilesSensor,
+ numberOfFileErrorsSensor);
}
@Test
public void shouldCorrectlyHandleHitRatioRecordingsWithZeroHitsAndMisses() {
- reset(statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- expect(statisticsToAdd1.getHistogramData(anyObject())).andStubReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
- expect(statisticsToAdd1.getAndResetTickerCount(anyObject())).andStubReturn(0L);
- replay(statisticsToAdd1);
- memtableHitRatioSensor.record(0, 0L);
- blockCacheDataHitRatioSensor.record(0, 0L);
- blockCacheIndexHitRatioSensor.record(0, 0L);
- blockCacheFilterHitRatioSensor.record(0, 0L);
- replay(memtableHitRatioSensor);
- replay(blockCacheDataHitRatioSensor);
- replay(blockCacheIndexHitRatioSensor);
- replay(blockCacheFilterHitRatioSensor);
+ when(statisticsToAdd1.getHistogramData(any())).thenReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
+ when(statisticsToAdd1.getAndResetTickerCount(any())).thenReturn(0L);
recorder.record(0L);
- verify(memtableHitRatioSensor);
- verify(blockCacheDataHitRatioSensor);
- verify(blockCacheIndexHitRatioSensor);
- verify(blockCacheFilterHitRatioSensor);
+ verify(memtableHitRatioSensor).record(0d, 0L);
+ verify(blockCacheDataHitRatioSensor).record(0d, 0L);
+ verify(blockCacheIndexHitRatioSensor).record(0d, 0L);
+ verify(blockCacheFilterHitRatioSensor).record(0d, 0L);
+ verify(statisticsToAdd1, times(2)).getHistogramData(any());
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(any());
}
@Test
public void shouldCorrectlyHandleAvgRecordingsWithZeroSumAndCount() {
- reset(statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- expect(statisticsToAdd1.getHistogramData(anyObject())).andStubReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
- expect(statisticsToAdd1.getAndResetTickerCount(anyObject())).andStubReturn(0L);
- replay(statisticsToAdd1);
- memtableAvgFlushTimeSensor.record(0, 0L);
- compactionTimeAvgSensor.record(0, 0L);
- replay(memtableAvgFlushTimeSensor);
- replay(compactionTimeAvgSensor);
+ final long now = 0L;
+ when(statisticsToAdd1.getHistogramData(any())).thenReturn(new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0L, 0L, 0.0));
+ when(statisticsToAdd1.getAndResetTickerCount(any())).thenReturn(0L);
- recorder.record(0L);
+ recorder.record(now);
- verify(memtableAvgFlushTimeSensor);
- verify(compactionTimeAvgSensor);
+ verify(compactionTimeAvgSensor).record(0d, now);
+ verify(memtableAvgFlushTimeSensor).record(0d, now);
+ verify(statisticsToAdd1, times(2)).getHistogramData(any());
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(any());
Review Comment:
Those verifications are not needed since you have already done them in `shouldRecordStatisticsBasedMetrics()`. This method tests the computation of the average and not the retrieval of the single values.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- verify(recordingTrigger);
+ verify(recordingTrigger, never()).removeMetricsRecorder(recorder);
- reset(recordingTrigger);
- recordingTrigger.removeMetricsRecorder(recorder);
- replay(recordingTrigger);
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_2);
- verify(recordingTrigger);
+ verify(recordingTrigger).removeMetricsRecorder(recorder);
}
@Test
public void shouldThrowIfValueProvidersToRemoveNotFound() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
assertThrows(
- IllegalStateException.class,
- () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
+ IllegalStateException.class,
+ () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
);
}
@Test
public void shouldRecordStatisticsBasedMetrics() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(statisticsToAdd1);
- reset(statisticsToAdd2);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(1L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(2L);
- bytesWrittenToDatabaseSensor.record(1 + 2, 0L);
- replay(bytesWrittenToDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(3L);
- bytesReadFromDatabaseSensor.record(2 + 3, 0L);
- replay(bytesReadFromDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(4L);
- memtableBytesFlushedSensor.record(3 + 4, 0L);
- replay(memtableBytesFlushedSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(1L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(4L);
- memtableHitRatioSensor.record((double) 4 / (4 + 6), 0L);
- replay(memtableHitRatioSensor);
+ final long now = 0L;
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(1L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(2L);
+ doNothing().when(bytesWrittenToDatabaseSensor).record(1 + 2, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(3L);
+ doNothing().when(bytesReadFromDatabaseSensor).record(2 + 3, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(memtableBytesFlushedSensor).record(3 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(1L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(4L);
+ doNothing().when(memtableHitRatioSensor).record((double) 4 / (4 + 6), now);
final HistogramData memtableFlushTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 10L, 3.0);
final HistogramData memtableFlushTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 20.0, 4L, 8L, 10.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData2);
- memtableAvgFlushTimeSensor.record((double) (10 + 8) / (2 + 4), 0L);
- replay(memtableAvgFlushTimeSensor);
- memtableMinFlushTimeSensor.record(3.0, 0L);
- replay(memtableMinFlushTimeSensor);
- memtableMaxFlushTimeSensor.record(20.0, 0L);
- replay(memtableMaxFlushTimeSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(5L);
- writeStallDurationSensor.record(4 + 5, 0L);
- replay(writeStallDurationSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(2L);
- blockCacheDataHitRatioSensor.record((double) 8 / (8 + 6), 0L);
- replay(blockCacheDataHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(4L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(4L);
- blockCacheIndexHitRatioSensor.record((double) 6 / (6 + 6), 0L);
- replay(blockCacheIndexHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(2L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(5L);
- blockCacheFilterHitRatioSensor.record((double) 5 / (5 + 9), 0L);
- replay(blockCacheFilterHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(4L);
- bytesWrittenDuringCompactionSensor.record(2 + 4, 0L);
- replay(bytesWrittenDuringCompactionSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(5L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(6L);
- bytesReadDuringCompactionSensor.record(5 + 6, 0L);
- replay(bytesReadDuringCompactionSensor);
+ when(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData2);
+ doNothing().when(memtableAvgFlushTimeSensor).record((double) (10 + 8) / (2 + 4), now);
+ doNothing().when(memtableMinFlushTimeSensor).record(3.0, now);
+ doNothing().when(memtableMaxFlushTimeSensor).record(20.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(5L);
+ doNothing().when(writeStallDurationSensor).record(4 + 5, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(2L);
+ doNothing().when(blockCacheDataHitRatioSensor).record((double) 8 / (8 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(4L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(4L);
+ doNothing().when(blockCacheIndexHitRatioSensor).record((double) 6 / (6 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(2L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(5L);
+ doNothing().when(blockCacheFilterHitRatioSensor).record((double) 5 / (5 + 9), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(bytesWrittenDuringCompactionSensor).record(2 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(5L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(6L);
+ doNothing().when(bytesReadDuringCompactionSensor).record(5 + 6, now);
final HistogramData compactionTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 8L, 6.0);
final HistogramData compactionTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 24.0, 2L, 8L, 4.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData2);
- compactionTimeAvgSensor.record((double) (8 + 8) / (2 + 2), 0L);
- replay(compactionTimeAvgSensor);
- compactionTimeMinSensor.record(4.0, 0L);
- replay(compactionTimeMinSensor);
- compactionTimeMaxSensor.record(24.0, 0L);
- replay(compactionTimeMaxSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(7L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(4L);
- numberOfOpenFilesSensor.record((5 + 7) - (3 + 4), 0L);
- replay(numberOfOpenFilesSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(34L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(11L);
- numberOfFileErrorsSensor.record(11 + 34, 0L);
- replay(numberOfFileErrorsSensor);
-
- replay(statisticsToAdd1);
- replay(statisticsToAdd2);
-
- recorder.record(0L);
-
- verify(statisticsToAdd1);
- verify(statisticsToAdd2);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ when(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData2);
+ doNothing().when(compactionTimeAvgSensor).record((double) (8 + 8) / (2 + 2), now);
+ doNothing().when(compactionTimeMinSensor).record(4.0, now);
+ doNothing().when(compactionTimeMaxSensor).record(24.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(7L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(4L);
+ doNothing().when(numberOfOpenFilesSensor).record((5 + 7) - (3 + 4), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(34L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(11L);
+ doNothing().when(numberOfFileErrorsSensor).record(11 + 34, now);
+
+ recorder.record(now);
+
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(bytesWrittenToDatabaseSensor).record(isA(Double.class), eq(now));
Review Comment:
This verification should be
```
verify(bytesWrittenToDatabaseSensor).record(eq(Double.valueOf(1 + 2)), eq(now));
```
Similar applies to the other verifications. Use the values of the void stubbings above you need to remove.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java:
##########
@@ -279,449 +310,335 @@ public void shouldThrowIfDbToAddWasAlreadyAddedForOtherSegment() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
final Throwable exception = assertThrows(
- IllegalStateException.class,
- () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
+ IllegalStateException.class,
+ () -> recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd1, cacheToAdd2, statisticsToAdd2)
);
assertThat(
- exception.getMessage(),
- is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
- " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
- "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
+ exception.getMessage(),
+ is("DB instance for store " + SEGMENT_STORE_NAME_2 + " of task " + TASK_ID1 +
+ " was already added for another segment as a value provider. This is a bug in Kafka Streams. " +
+ "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues")
);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedToNewlyCreatedRecorder() {
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
-
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldAddItselfToRecordingTriggerWhenFirstValueProvidersAreAddedAfterLastValueProvidersWereRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- reset(recordingTrigger);
- recordingTrigger.addMetricsRecorder(recorder);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verify(recordingTrigger).addMetricsRecorder(recorder);
}
@Test
public void shouldNotAddItselfToRecordingTriggerWhenNotEmpty2() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ verify(recordingTrigger).addMetricsRecorder(recorder);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- verify(recordingTrigger);
+ verifyNoMoreInteractions(recordingTrigger);
}
@Test
public void shouldCloseStatisticsWhenValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
- reset(statisticsToAdd1);
- statisticsToAdd1.close();
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1).close();
}
@Test
public void shouldNotCloseStatisticsWhenValueProvidersWithoutStatisticsAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, null);
- reset(statisticsToAdd1);
- replay(statisticsToAdd1);
-
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
-
- verify(statisticsToAdd1);
+ verify(statisticsToAdd1, never()).close();
}
@Test
public void shouldRemoveItselfFromRecordingTriggerWhenLastValueProvidersAreRemoved() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(recordingTrigger);
- replay(recordingTrigger);
+
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_1);
- verify(recordingTrigger);
+ verify(recordingTrigger, never()).removeMetricsRecorder(recorder);
- reset(recordingTrigger);
- recordingTrigger.removeMetricsRecorder(recorder);
- replay(recordingTrigger);
+ Mockito.reset(recordingTrigger);
recorder.removeValueProviders(SEGMENT_STORE_NAME_2);
- verify(recordingTrigger);
+ verify(recordingTrigger).removeMetricsRecorder(recorder);
}
@Test
public void shouldThrowIfValueProvidersToRemoveNotFound() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
assertThrows(
- IllegalStateException.class,
- () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
+ IllegalStateException.class,
+ () -> recorder.removeValueProviders(SEGMENT_STORE_NAME_2)
);
}
@Test
public void shouldRecordStatisticsBasedMetrics() {
recorder.addValueProviders(SEGMENT_STORE_NAME_1, dbToAdd1, cacheToAdd1, statisticsToAdd1);
recorder.addValueProviders(SEGMENT_STORE_NAME_2, dbToAdd2, cacheToAdd2, statisticsToAdd2);
- reset(statisticsToAdd1);
- reset(statisticsToAdd2);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(1L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).andReturn(2L);
- bytesWrittenToDatabaseSensor.record(1 + 2, 0L);
- replay(bytesWrittenToDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).andReturn(3L);
- bytesReadFromDatabaseSensor.record(2 + 3, 0L);
- replay(bytesReadFromDatabaseSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).andReturn(4L);
- memtableBytesFlushedSensor.record(3 + 4, 0L);
- replay(memtableBytesFlushedSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(1L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).andReturn(4L);
- memtableHitRatioSensor.record((double) 4 / (4 + 6), 0L);
- replay(memtableHitRatioSensor);
+ final long now = 0L;
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(1L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_WRITTEN)).thenReturn(2L);
+ doNothing().when(bytesWrittenToDatabaseSensor).record(1 + 2, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BYTES_READ)).thenReturn(3L);
+ doNothing().when(bytesReadFromDatabaseSensor).record(2 + 3, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.FLUSH_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(memtableBytesFlushedSensor).record(3 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(1L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.MEMTABLE_MISS)).thenReturn(4L);
+ doNothing().when(memtableHitRatioSensor).record((double) 4 / (4 + 6), now);
final HistogramData memtableFlushTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 10L, 3.0);
final HistogramData memtableFlushTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 20.0, 4L, 8L, 10.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).andReturn(memtableFlushTimeData2);
- memtableAvgFlushTimeSensor.record((double) (10 + 8) / (2 + 4), 0L);
- replay(memtableAvgFlushTimeSensor);
- memtableMinFlushTimeSensor.record(3.0, 0L);
- replay(memtableMinFlushTimeSensor);
- memtableMaxFlushTimeSensor.record(20.0, 0L);
- replay(memtableMaxFlushTimeSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).andReturn(5L);
- writeStallDurationSensor.record(4 + 5, 0L);
- replay(writeStallDurationSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).andReturn(2L);
- blockCacheDataHitRatioSensor.record((double) 8 / (8 + 6), 0L);
- replay(blockCacheDataHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(4L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).andReturn(4L);
- blockCacheIndexHitRatioSensor.record((double) 6 / (6 + 6), 0L);
- replay(blockCacheIndexHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(2L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(4L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).andReturn(5L);
- blockCacheFilterHitRatioSensor.record((double) 5 / (5 + 9), 0L);
- replay(blockCacheFilterHitRatioSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(2L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).andReturn(4L);
- bytesWrittenDuringCompactionSensor.record(2 + 4, 0L);
- replay(bytesWrittenDuringCompactionSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(5L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).andReturn(6L);
- bytesReadDuringCompactionSensor.record(5 + 6, 0L);
- replay(bytesReadDuringCompactionSensor);
+ when(statisticsToAdd1.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.FLUSH_TIME)).thenReturn(memtableFlushTimeData2);
+ doNothing().when(memtableAvgFlushTimeSensor).record((double) (10 + 8) / (2 + 4), now);
+ doNothing().when(memtableMinFlushTimeSensor).record(3.0, now);
+ doNothing().when(memtableMaxFlushTimeSensor).record(20.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.STALL_MICROS)).thenReturn(5L);
+ doNothing().when(writeStallDurationSensor).record(4 + 5, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_DATA_MISS)).thenReturn(2L);
+ doNothing().when(blockCacheDataHitRatioSensor).record((double) 8 / (8 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(4L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_HIT)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_INDEX_MISS)).thenReturn(4L);
+ doNothing().when(blockCacheIndexHitRatioSensor).record((double) 6 / (6 + 6), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(2L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(4L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_HIT)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.BLOCK_CACHE_FILTER_MISS)).thenReturn(5L);
+ doNothing().when(blockCacheFilterHitRatioSensor).record((double) 5 / (5 + 9), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(2L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_WRITE_BYTES)).thenReturn(4L);
+ doNothing().when(bytesWrittenDuringCompactionSensor).record(2 + 4, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(5L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.COMPACT_READ_BYTES)).thenReturn(6L);
+ doNothing().when(bytesReadDuringCompactionSensor).record(5 + 6, now);
final HistogramData compactionTimeData1 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 16.0, 2L, 8L, 6.0);
final HistogramData compactionTimeData2 = new HistogramData(0.0, 0.0, 0.0, 0.0, 0.0, 24.0, 2L, 8L, 4.0);
- expect(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData1);
- expect(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).andReturn(compactionTimeData2);
- compactionTimeAvgSensor.record((double) (8 + 8) / (2 + 2), 0L);
- replay(compactionTimeAvgSensor);
- compactionTimeMinSensor.record(4.0, 0L);
- replay(compactionTimeMinSensor);
- compactionTimeMaxSensor.record(24.0, 0L);
- replay(compactionTimeMaxSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(5L);
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(3L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).andReturn(7L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).andReturn(4L);
- numberOfOpenFilesSensor.record((5 + 7) - (3 + 4), 0L);
- replay(numberOfOpenFilesSensor);
-
- expect(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(34L);
- expect(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).andReturn(11L);
- numberOfFileErrorsSensor.record(11 + 34, 0L);
- replay(numberOfFileErrorsSensor);
-
- replay(statisticsToAdd1);
- replay(statisticsToAdd2);
-
- recorder.record(0L);
-
- verify(statisticsToAdd1);
- verify(statisticsToAdd2);
- verify(
- bytesWrittenToDatabaseSensor,
- bytesReadFromDatabaseSensor,
- memtableBytesFlushedSensor,
- memtableHitRatioSensor,
- memtableAvgFlushTimeSensor,
- memtableMinFlushTimeSensor,
- memtableMaxFlushTimeSensor,
- writeStallDurationSensor,
- blockCacheDataHitRatioSensor,
- blockCacheIndexHitRatioSensor,
- blockCacheFilterHitRatioSensor,
- bytesWrittenDuringCompactionSensor,
- bytesReadDuringCompactionSensor,
- compactionTimeAvgSensor,
- compactionTimeMinSensor,
- compactionTimeMaxSensor,
- numberOfOpenFilesSensor,
- numberOfFileErrorsSensor
- );
+ when(statisticsToAdd1.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData1);
+ when(statisticsToAdd2.getHistogramData(HistogramType.COMPACTION_TIME)).thenReturn(compactionTimeData2);
+ doNothing().when(compactionTimeAvgSensor).record((double) (8 + 8) / (2 + 2), now);
+ doNothing().when(compactionTimeMinSensor).record(4.0, now);
+ doNothing().when(compactionTimeMaxSensor).record(24.0, now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(5L);
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(3L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_OPENS)).thenReturn(7L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_CLOSES)).thenReturn(4L);
+ doNothing().when(numberOfOpenFilesSensor).record((5 + 7) - (3 + 4), now);
+
+ when(statisticsToAdd1.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(34L);
+ when(statisticsToAdd2.getAndResetTickerCount(TickerType.NO_FILE_ERRORS)).thenReturn(11L);
+ doNothing().when(numberOfFileErrorsSensor).record(11 + 34, now);
+
+ recorder.record(now);
+
+ verify(statisticsToAdd1, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(17)).getAndResetTickerCount(isA(TickerType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
+ verify(statisticsToAdd2, times(2)).getHistogramData(isA(HistogramType.class));
Review Comment:
You need to verify each `TickerType` and each `HistogramType` here. Using `isA()` is to general. I think the best thing would be to split this test method in one test method per computed metric.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org