Skip to content

KAFKA-20750: Add divide-by-zero error check in KafkaConsumerMetrics/KafkaShareConsumerMetrics. #22718

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

Merged
AndrewJSchofield merged 2 commits into from
Jul 1, 2026
+43 −2
Merged

KAFKA-20750: Add divide-by-zero error check in KafkaConsumerMetrics/KafkaShareConsumerMetrics. #22718

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7ac37e7
Add divide-by-zero error check
ShivsundarR Jul 1, 2026
fbe3dd3
Nit comments
ShivsundarR Jul 1, 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
3 changes: 2 additions & 1 deletion ...c/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,8 @@ public void recordPollStart(long pollStartMs) {

public void recordPollEnd(long pollEndMs) {
long pollTimeMs = pollEndMs - pollStartMs;
double pollIdleRatio = pollTimeMs * 1.0 / (pollTimeMs + timeSinceLastPollMs);
long pollCycleTimeMs = pollTimeMs + timeSinceLastPollMs;
double pollIdleRatio = pollCycleTimeMs == 0 ? 0.0 : (pollTimeMs * 1.0 / pollCycleTimeMs);
this.pollIdleSensor.record(pollIdleRatio);
}

Expand Down
3 changes: 2 additions & 1 deletion ...n/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaShareConsumerMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,8 @@ public void recordPollStart(long pollStartMs) {

public void recordPollEnd(long pollEndMs) {
long pollTimeMs = pollEndMs - pollStartMs;
double pollIdleRatio = pollTimeMs * 1.0 / (pollTimeMs + timeSinceLastPollMs);
long pollCycleTimeMs = pollTimeMs + timeSinceLastPollMs;
double pollIdleRatio = pollCycleTimeMs == 0 ? 0.0 : (pollTimeMs * 1.0 / pollCycleTimeMs);
this.pollIdleSensor.record(pollIdleRatio);
}
}
20 changes: 20 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3775,6 +3775,26 @@ public void testPollIdleRatio(GroupProtocol groupProtocol) {
assertEquals((1.0d + 0.0d + 0.5d) / 3, consumer.metrics().get(pollIdleRatio).metricValue());
}

@ParameterizedTest
@EnumSource(GroupProtocol.class)
public void testPollIdleRatioZero(GroupProtocol groupProtocol) {
ConsumerMetadata metadata = createMetadata(subscription);
MockClient client = new MockClient(time, metadata);
initMetadata(client, Map.of(topic, 1));

KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
// MetricName object to check
Metrics metrics = consumer.metricsRegistry();
MetricName pollIdleRatio = metrics.metricName("poll-idle-ratio-avg", "consumer-metrics");
// Test default value
assertEquals(Double.NaN, consumer.metrics().get(pollIdleRatio).metricValue());

// Poll starts and ends within the same millisecond, so the metric should be 0.
consumer.kafkaConsumerMetrics().recordPollStart(time.milliseconds());
consumer.kafkaConsumerMetrics().recordPollEnd(time.milliseconds());
assertEquals(0.0d, consumer.metrics().get(pollIdleRatio).metricValue());
}

private static boolean consumerMetricPresent(KafkaConsumer<String, String> consumer, String name) {
MetricName metricName = new MetricName(name, "consumer-metrics", "", Collections.emptyMap());
return consumer.metricsRegistry().metrics().containsKey(metricName);
Expand Down
19 changes: 19 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,25 @@ public void testPollIdleRatio() {
assertEquals((1.0d + 0.0d + 0.5d) / 3, consumer.metrics().get(pollIdleRatio).metricValue());
}

@Test
public void testPollIdleRatioZero() {
ShareConsumerMetadata metadata = createMetadata(subscription);
MockClient client = new MockClient(time, metadata);
initMetadata(client, Map.of(topic, 1));

KafkaShareConsumer<String, String> consumer = newShareConsumer(time, client, subscription, metadata);
// MetricName object to check
Metrics metrics = consumer.metricsRegistry();
MetricName pollIdleRatio = metrics.metricName("poll-idle-ratio-avg", CONSUMER_SHARE_METRIC_GROUP_PREFIX + "-metrics");
// Test default value
assertEquals(Double.NaN, consumer.metrics().get(pollIdleRatio).metricValue());

// Poll starts and ends within the same millisecond, so the metric should be 0.
consumer.kafkaShareConsumerMetrics().recordPollStart(time.milliseconds());
consumer.kafkaShareConsumerMetrics().recordPollEnd(time.milliseconds());
assertEquals(0.0d, consumer.metrics().get(pollIdleRatio).metricValue());
}

private static boolean consumerMetricPresent(KafkaShareConsumer<String, String> consumer, String name) {
MetricName metricName = new MetricName(name, CONSUMER_SHARE_METRIC_GROUP_PREFIX + "-metrics", "", Map.of());
return consumer.metricsRegistry().metrics().containsKey(metricName);
Expand Down
Loading