Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mjsax wants to merge 4 commits into
base: trunk
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1c7d814
KAFKA-20877: Fix incorrect restore-rate and update-rate metrics
Jul 31, 2026
32f699b
review comments
Aug 3, 2026
4e4ca39
align code style
Aug 4, 2026
1c6552f
more cleanup
Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 15 additions & 14 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -189,7 +189,7 @@ public static Sensor restoreSensor(final String threadId,
final String taskId,
final StreamsMetricsImpl streamsMetrics,
final Sensor... parentSensor) {
return invocationRateAndTotalSensor(
return rateAndTotalSensor(

@aliehsaeedii aliehsaeedii Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I assume you changed the code so that it matches with what we have in docs/operations/monitoring.md ? So no need to update the doc? Should we mention this somewhere so that users understand that the change in number is due to an improvement in how we measure it, rather than a change in their application?

@mjsax mjsax Aug 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Don't think we need to call it out -- that is what the release changelog is for, and KAFKA-20877 will be listed.

That's actually the main reason, why I did not do this a MINOR PR, but filed a Jira, to highlight the fix.

threadId,
taskId,
RESTORE,
Expand All @@ -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,
Expand Down Expand Up @@ -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(

@aliehsaeedii aliehsaeedii Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This changes dropped-records-rate too: almost all call sites record 1 per record, but InMemoryWindowStore:158 and InMemorySessionStore:163 record a whole batch (expiredRecordSensor.record(expiredRecords, ...))

@mjsax mjsax Aug 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh. So this metric (dropped-records-rate) is actually broken for InMemoryWindowStore and InMemorySessionStore, too. Great catch.

Guess we just need to update the Jira and PR description, as this PR does fix the issue, too?

@mjsax mjsax Aug 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, it seems the issue for InMemoryWindowStore and InMemorySessionStore was only introduced recently on trunk only: d2328ff

So we don't even need to update the Jira description as there is no issue in 4.3 and older releases.

threadId,
taskId,
DROPPED_RECORDS,
Expand Down Expand Up @@ -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;
}

Expand Down
20 changes: 17 additions & 3 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,9 +75,9 @@
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.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
Expand Down Expand Up @@ -487,12 +487,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
assertTrue(
// regression test for KAFKA-20877: previously we did incorrectly count batches which would result in 0.03333
// using 0.5 as good intermediate to the expected value of 0.83333
// -> avoid equalTo(...) on floating point numbers
0.5d < ((Number) rateMetric.metricValue()).doubleValue(),

@aliehsaeedii aliehsaeedii Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as in StreamTaskTest

"Expected a value larger 0.5 [precisely 0.83333...], but got " + rateMetric.metricValue()
);

task.recordRestoration(time, 50L, false);

assertThat(totalMetric.metricValue(), equalTo(75.0));
assertThat(rateMetric.metricValue(), not(0.0));
assertTrue(
// regression test for KAFKA-20877: previously we did incorrectly count batches which would result in 0.06666
// using 2.0 as good intermediate to the expected value of 2.5
// -> avoid equalTo(...) on floating point numbers

@aliehsaeedii aliehsaeedii Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: So could we use the JUnit's three-arg assertEquals(expected, actual, delta)?

2.0d < ((Number) rateMetric.metricValue()).doubleValue(),
"Expected a value larger 2.0 [precisely 2.5], but got " + rateMetric.metricValue()
);
}

private KafkaMetric getMetric(final String operation,
Expand Down
18 changes: 16 additions & 2 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
assertTrue(
// regression test for KAFKA-20877: previously we did incorrectly count batches which would result in 0.03333
// using 0.5 as good intermediate to the expected value of 0.83333
// -> avoid equalTo(...) on floating point numbers
0.5d < ((Number) rateMetric.metricValue()).doubleValue(),
"Expected a value larger 0.5 [precisely 0.83333...], but got " + rateMetric.metricValue()
);
assertThat(remainMetric.metricValue(), equalTo(75.0));

task.recordRestoration(time, 50L, false);

assertThat(totalMetric.metricValue(), equalTo(75.0));
assertThat(rateMetric.metricValue(), not(0.0));
assertTrue(
// regression test for KAFKA-20877: previously we did incorrectly count batches which would result in 0.06666
// using 2.0 as good intermediate to the expected value of 2.5
// -> avoid equalTo(...) on floating point numbers
2.0d < ((Number) rateMetric.metricValue()).doubleValue(),
"Expected a value larger 2.0 [precisely 2.5], but got " + rateMetric.metricValue()
);
assertThat(remainMetric.metricValue(), equalTo(25.0));
}

Expand Down
13 changes: 2 additions & 11 deletions ...s/src/test/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
);
Expand Down
Loading