KAFKA-20877: Fix incorrect restore-rate and update-rate metrics

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -2229,7 +2229,7 @@ private void recordRatio(final long now, final WindowedSum windowedSum, final Se
         if (runOnceLatencyWindow > 0.0) {
             final double latencyWindow =
                 windowedSum.measure(metricsConfig, now);
-            ratioSensor.record(latencyWindow / runOnceLatencyWindow);
+            ratioSensor.record(latencyWindow / runOnceLatencyWindow, now);
         } else {
             ratioSensor.record(0.0, now);
         }

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java

@@ -28,7 +28,7 @@
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TOTAL_SUFFIX;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndMaxToSensor;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCountToSensor;
-import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateToSensor;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addRateOfSumAndSumMetricsToSensor;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addSumMetricToSensor;
 import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addValueMetricToSensor;
 
@@ -189,7 +189,7 @@ public static Sensor restoreSensor(final String threadId,
                                        final String taskId,
                                        final StreamsMetricsImpl streamsMetrics,
                                        final Sensor... parentSensor) {
-        return invocationRateAndTotalSensor(
+        return rateAndTotalSensor(
             threadId,
             taskId,
             RESTORE,
@@ -205,7 +205,7 @@ public static Sensor updateSensor(final String threadId,
                                       final String taskId,
                                       final StreamsMetricsImpl streamsMetrics,
                                       final Sensor... parentSensor) {
-        return invocationRateAndTotalSensor(
+        return rateAndTotalSensor(
             threadId,
             taskId,
             UPDATE,
@@ -250,7 +250,7 @@ public static Sensor recordLatenessSensor(final String threadId,
     public static Sensor droppedRecordsSensor(final String threadId,
                                               final String taskId,
                                               final StreamsMetricsImpl streamsMetrics) {
-        return invocationRateAndTotalSensor(
+        return rateAndTotalSensor(
             threadId,
             taskId,
             DROPPED_RECORDS,
@@ -281,19 +281,20 @@ private static Sensor invocationRateAndCountSensor(final String threadId,
         return sensor;
     }
 
-    private static Sensor invocationRateAndTotalSensor(final String threadId,
-                                                       final String taskId,
-                                                       final String operation,
-                                                       final String descriptionOfRate,
-                                                       final String descriptionOfTotal,
-                                                       final RecordingLevel recordingLevel,
-                                                       final StreamsMetricsImpl streamsMetrics,
-                                                       final Sensor... parentSensors) {
+    private static Sensor rateAndTotalSensor(
+        final String threadId,
+        final String taskId,
+        final String operation,
+        final String descriptionOfRate,
+        final String descriptionOfTotal,
+        final RecordingLevel recordingLevel,
+        final StreamsMetricsImpl streamsMetrics,
+        final Sensor... parentSensors
+    ) {
         final Sensor sensor = streamsMetrics.taskLevelSensor(threadId, taskId, operation, recordingLevel, parentSensors);
         final Map<String, String> tags = streamsMetrics.taskLevelTagMap(threadId, taskId);
 
-        addInvocationRateToSensor(sensor, TASK_LEVEL_GROUP, tags, operation, descriptionOfRate);
-        addSumMetricToSensor(sensor, TASK_LEVEL_GROUP, tags, operation, true, descriptionOfTotal);
+        addRateOfSumAndSumMetricsToSensor(sensor, TASK_LEVEL_GROUP, tags, operation, descriptionOfRate, descriptionOfTotal);
         return sensor;
     }
 

streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStore.java

@@ -189,7 +189,7 @@ public void put(final Bytes rawBaseKey,
         final S segment = segments.getOrCreateSegmentIfLive(segmentId, internalProcessorContext, observedStreamTime);
 
         if (segment == null) {
-            expiredRecordSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());
+            expiredRecordSensor.record();
             LOG.warn("Skipping record for expired segment.");
         } else {
             synchronized (position) {

streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java

@@ -254,7 +254,7 @@ public void put(final Bytes key,
         final long segmentId = segments.segmentId(timestamp);
         final S segment = segments.getOrCreateSegmentIfLive(segmentId, internalProcessorContext, observedStreamTime);
         if (segment == null) {
-            expiredRecordSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());
+            expiredRecordSensor.record();
         } else {
             synchronized (position) {
                 // Transactional puts stage their position too, so READ_COMMITTED never sees a position ahead of the data.

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java

@@ -159,8 +159,8 @@ public void init(final StateStoreContext stateStoreContext,
                         }
                         removeExpiredSegments();
                         if (expiredRecords > 0) {
-                            if (expiredRecordSensor != null && context != null) {
-                                expiredRecordSensor.record(expiredRecords, context.currentSystemTimeMs());
+                            if (expiredRecordSensor != null) {
+                                expiredRecordSensor.record(expiredRecords);
                             }
                             LOG.warn("Skipping {} records for expired segments.", expiredRecords);
                         }
@@ -205,8 +205,8 @@ public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
             if (windowEndTimestamp <= observedStreamTime - retentionPeriod) {
                 // The provided context is not required to implement InternalProcessorContext,
                 // If it doesn't, we can't record this metric (in fact, we wouldn't have even initialized it).
-                if (expiredRecordSensor != null && context != null) {
-                    expiredRecordSensor.record(1.0d, context.currentSystemTimeMs());
+                if (expiredRecordSensor != null) {
+                    expiredRecordSensor.record();
                 }
                 LOG.warn("Skipping record for expired segment.");
             } else if (transactionBuffer != null) {

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java

@@ -155,7 +155,7 @@ public void init(final StateStoreContext stateStoreContext,
                         }
                         removeExpiredSegments();
                         if (expiredRecords > 0) {
-                            expiredRecordSensor.record(expiredRecords, internalProcessorContext.currentSystemTimeMs());
+                            expiredRecordSensor.record(expiredRecords);
                             LOG.warn("Skipping {} records for expired segments.", expiredRecords);
                         }
                     }
@@ -195,7 +195,7 @@ public void put(final Bytes key, final byte[] value, final long windowStartTimes
 
         synchronized (position) {
             if (windowStartTimestamp <= observedStreamTime - retentionPeriod) {
-                expiredRecordSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());
+                expiredRecordSensor.record();
                 LOG.warn("Skipping record for expired segment.");
             } else if (transactionBuffer != null) {
                 if (value != null) {

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java

@@ -132,7 +132,7 @@ public long put(final Bytes key, final byte[] value, final long timestamp) {
 
         synchronized (position) {
             if (timestamp < observedStreamTime - gracePeriod) {
-                expiredRecordSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());
+                expiredRecordSensor.record();
                 LOG.warn("Skipping record for expired put.");
                 StoreQueryUtils.updatePosition(position, internalProcessorContext);
                 return PUT_RETURN_CODE_NOT_PUT;
@@ -160,7 +160,7 @@ public VersionedRecord<byte[]> delete(final Bytes key, final long timestamp) {
 
         synchronized (position) {
             if (timestamp < observedStreamTime - gracePeriod) {
-                expiredRecordSensor.record(1.0d, internalProcessorContext.currentSystemTimeMs());
+                expiredRecordSensor.record();
                 LOG.warn("Skipping record for expired delete.");
                 return null;
             }

streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java

@@ -75,7 +75,6 @@
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.isA;
-import static org.hamcrest.Matchers.not;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
@@ -487,12 +486,26 @@ public void shouldRecordRestoredRecords() {
         task.recordRestoration(time, 25L, false);
 
         assertThat(totalMetric.metricValue(), equalTo(25.0));
-        assertThat(rateMetric.metricValue(), not(0.0));
+        // the rate measures updated records per second, not update batches per second; with no time
+        // elapsed the rate window is (metrics.num.samples - 1) * metrics.sample.window.ms == 30s
+        assertEquals(
+            25.0 / 30.0,
+            ((Number) rateMetric.metricValue()).doubleValue(),
+            0.0001d,
+            "update-rate must measure updated records per second, not update batches per second; "
+                + "counting batches would give 1/30 == 0.03333 (KAFKA-20877)"
+        );
 
         task.recordRestoration(time, 50L, false);
 
         assertThat(totalMetric.metricValue(), equalTo(75.0));
-        assertThat(rateMetric.metricValue(), not(0.0));
+        assertEquals(
+            75.0 / 30.0,
+            ((Number) rateMetric.metricValue()).doubleValue(),
+            0.0001d,
+            "update-rate must measure updated records per second, not update batches per second; "
+                + "counting batches would give 2/30 == 0.06666 (KAFKA-20877)"
+        );
     }
 
     private KafkaMetric getMetric(final String operation,

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java

@@ -893,13 +893,27 @@ public void shouldRecordRestoredRecords() {
         task.recordRestoration(time, 25L, false);
 
         assertThat(totalMetric.metricValue(), equalTo(25.0));
-        assertThat(rateMetric.metricValue(), not(0.0));
+        // the rate measures restored records per second, not restore batches per second; with no time
+        // elapsed the rate window is (metrics.num.samples - 1) * metrics.sample.window.ms == 30s
+        assertEquals(
+            25.0 / 30.0,
+            ((Number) rateMetric.metricValue()).doubleValue(),
+            0.0001d,
+            "restore-rate must measure restored records per second, not restore batches per second; "
+                + "counting batches would give 1/30 == 0.03333 (KAFKA-20877)"
+        );
         assertThat(remainMetric.metricValue(), equalTo(75.0));
 
         task.recordRestoration(time, 50L, false);
 
         assertThat(totalMetric.metricValue(), equalTo(75.0));
-        assertThat(rateMetric.metricValue(), not(0.0));
+        assertEquals(
+            75.0 / 30.0,
+            ((Number) rateMetric.metricValue()).doubleValue(),
+            0.0001d,
+            "restore-rate must measure restored records per second, not restore batches per second; "
+                + "counting batches would give 2/30 == 0.06666 (KAFKA-20877)"
+        );
         assertThat(remainMetric.metricValue(), equalTo(25.0));
     }
 

streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetricsTest.java

@@ -240,21 +240,12 @@ public void shouldGetDroppedRecordsSensor() {
         try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
             final Sensor sensor = TaskMetrics.droppedRecordsSensor(THREAD_ID, TASK_ID, streamsMetrics);
             streamsMetricsStaticMock.verify(
-                () -> StreamsMetricsImpl.addInvocationRateToSensor(
+                () -> StreamsMetricsImpl.addRateOfSumAndSumMetricsToSensor(
                     expectedSensor,
                     TASK_LEVEL_GROUP,
                     tagMap,
                     operation,
-                    rateDescription
-                )
-            );
-            streamsMetricsStaticMock.verify(
-                () -> StreamsMetricsImpl.addSumMetricToSensor(
-                    expectedSensor,
-                    TASK_LEVEL_GROUP,
-                    tagMap,
-                    operation,
-                    true,
+                    rateDescription,
                     totalDescription
                 )
             );

streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java

@@ -88,7 +88,6 @@
 import static org.hamcrest.Matchers.hasEntry;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -1586,8 +1585,6 @@ public void shouldMeasureExpiredRecords() {
             TestUtils.tempDirectory(),
             new StreamsConfig(streamsConfig)
         );
-        final Time time = Time.SYSTEM;
-        context.setSystemTimeMs(time.milliseconds());
         bytesStore.init(context, bytesStore);
 
         // write a record to advance stream time, with a high enough timestamp
@@ -1622,7 +1619,16 @@ public void shouldMeasureExpiredRecords() {
             )
         ));
         assertEquals(1.0, dropTotal.metricValue());
-        assertNotEquals(0.0, dropRate.metricValue());
+        // exactly one record was dropped, over the rate's default un-elapsed sampling window of
+        // (metrics.num.samples - 1) * metrics.sample.window.ms == 30s. The delta is generous because the
+        // window also grows by however long the store work takes between recording and reading the
+        // metric; it still separates one dropped record from none (0.0) and from two (0.06666).
+        assertEquals(
+            1.0 / 30.0,
+            ((Number) dropRate.metricValue()).doubleValue(),
+            0.005d,
+            "dropped-records-rate should reflect the single dropped record over the ~30s sampling window"
+        );
 
         bytesStore.close();
     }

streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStoreTest.java

@@ -95,7 +95,6 @@
 import static org.hamcrest.Matchers.hasEntry;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -962,8 +961,6 @@ public void shouldMeasureExpiredRecords(final SegmentedBytesStore.KeySchema sche
             TestUtils.tempDirectory(),
             new StreamsConfig(streamsConfig)
         );
-        final Time time = Time.SYSTEM;
-        context.setSystemTimeMs(time.milliseconds());
         bytesStore.init(context, bytesStore);
 
         // write a record to advance stream time, with a high enough timestamp
@@ -998,7 +995,16 @@ public void shouldMeasureExpiredRecords(final SegmentedBytesStore.KeySchema sche
             )
         ));
         assertEquals(1.0, dropTotal.metricValue());
-        assertNotEquals(0.0, dropRate.metricValue());
+        // exactly one record was dropped, over the rate's default un-elapsed sampling window of
+        // (metrics.num.samples - 1) * metrics.sample.window.ms == 30s. The delta is generous because the
+        // window also grows by however long the store work takes between recording and reading the
+        // metric; it still separates one dropped record from none (0.0) and from two (0.06666).
+        assertEquals(
+            1.0 / 30.0,
+            ((Number) dropRate.metricValue()).doubleValue(),
+            0.005d,
+            "dropped-records-rate should reflect the single dropped record over the ~30s sampling window"
+        );
 
         bytesStore.close();
     }

streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java

@@ -71,7 +71,6 @@
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -894,9 +893,7 @@ public void shouldMeasureExpiredRecords() {
             new StreamsConfig(streamsConfig),
             recordCollector
         );
-        final Time time = Time.SYSTEM;
         context.setTime(1L);
-        context.setSystemTimeMs(time.milliseconds());
         sessionStore.init(context, sessionStore);
 
         // Advance stream time by inserting record with large enough timestamp that records with timestamp 0 are expired
@@ -931,7 +928,16 @@ public void shouldMeasureExpiredRecords() {
             )
         ));
         assertEquals(1.0, dropTotal.metricValue());
-        assertNotEquals(0.0, dropRate.metricValue());
+        // exactly one record was dropped, over the rate's default un-elapsed sampling window of
+        // (metrics.num.samples - 1) * metrics.sample.window.ms == 30s. The delta is generous because the
+        // window also grows by however long the store work takes between recording and reading the
+        // metric; it still separates one dropped record from none (0.0) and from two (0.06666).
+        assertEquals(
+            1.0 / 30.0,
+            ((Number) dropRate.metricValue()).doubleValue(),
+            0.005d,
+            "dropped-records-rate should reflect the single dropped record over the ~30s sampling window"
+        );
 
         sessionStore.close();
     }