aliehsaeedii commented on code in PR #23009:
URL: https://github.com/apache/kafka/pull/23009#discussion_r3694029222
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java:
##########
@@ -189,7 +189,7 @@ public static Sensor restoreSensor(final String threadId,
final String taskId,
final StreamsMetricsImpl streamsMetrics,
final Sensor... parentSensor) {
- return invocationRateAndTotalSensor(
+ return rateAndTotalSensor(
Review Comment:
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?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java:
##########
@@ -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(
Review Comment:
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, ...)`)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java:
##########
@@ -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(),
Review Comment:
Same as in `StreamTaskTest`
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java:
##########
@@ -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(),
+ "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
Review Comment:
nit: So could we use the JUnit's three-arg `assertEquals(expected, actual,
delta)`?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
