You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2024/12/17 15:17:16 UTC
[PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
apoorvmittal10 opened a new pull request, #18232:
URL: https://github.com/apache/kafka/pull/18232
The PR implements the behaviour defined in KIP-1109. It corrects the consumer topic and topic-partition metrics, while deprecating the incorrect ones.
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1109%3A+Unifying+Kafka+Consumer+Topic+Metrics
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900424589
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +195,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
Review Comment:
Generally speaking, we should always include a note explaining the deprecation. This should include the first version it was deprecated, when it's expected to be removed and what should be used instead.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900425014
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
Hmm, if we want insertion ordering to be preserved, we should change `SensorBuilder` to use a `LinkedHashMap` (`SequencedMap` would be better, but that requires Java 21). Having this implicit requirement and hoping people get it right is very error prone.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900385984
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
@ijuma please see my previous comment https://github.com/apache/kafka/pull/18232#discussion_r1889943665
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900771421
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
Yeah agree a explicit requirement to send an ordered map would be better in builder itself.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1892607155
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -162,16 +170,22 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
metrics.removeSensor(partitionRecordsLagMetricName(tp));
metrics.removeSensor(partitionRecordsLeadMetricName(tp));
metrics.removeMetric(partitionPreferredReadReplicaMetricName(tp));
+ // Remove deprecated metrics.
+ metrics.removeSensor(deprecatedMetricName(partitionRecordsLagMetricName(tp)));
+ metrics.removeSensor(deprecatedMetricName(partitionRecordsLeadMetricName(tp)));
+ metrics.removeMetric(partitionPreferredReadReplicaMetricNameDeprecated(tp));
Review Comment:
Done.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -117,20 +119,40 @@ public void testBytesFetched() {
@Test
public void testBytesFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordBytesFetched(topicName1, 2);
+ // 4 new metrics shall be registered.
+ assertEquals(4, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordBytesFetched(topicName2, 1);
+ // Another 8 metrics gets registered as deprecated metrics should be reported for topicName2.
Review Comment:
Done.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -145,46 +167,209 @@ public void testRecordsFetched() {
@Test
public void testRecordsFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordRecordsFetched(topicName1, 2);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordRecordsFetched(topicName2, 1);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for topicName2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
time.sleep(metrics.config().timeWindowMs() + 1);
metricsManager.recordRecordsFetched(topicName1, 10);
metricsManager.recordRecordsFetched(topicName2, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for topicName1.
assertEquals(6, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags1), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags1) > 0);
+ assertEquals(12, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags1), EPSILON);
+ // Validate metrics for topicName2.
assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags2), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags2) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, deprecatedTags), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, deprecatedTags) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLag() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLag(TP, 14);
- metricsManager.recordPartitionLag(TP, 8);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLag(tp1, 8);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLag(TP, 5);
+ metricsManager.recordPartitionLag(tp1, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(14, metricValue(metricsRegistry.recordsLagMax), EPSILON);
- assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags), EPSILON);
- assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags), EPSILON);
- assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags), EPSILON);
+ assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags1), EPSILON);
+ assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags1), EPSILON);
+ assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLag(tp2, 7);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ metricsManager.recordPartitionLag(tp2, 3);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLag(tp2, 2);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(7, metricValue(metricsRegistry.recordsLagMax), EPSILON);
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, tags2), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, tags2), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, deprecatedTags), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, deprecatedTags), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLead() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLead(TP, 15);
- metricsManager.recordPartitionLead(TP, 11);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLead(tp1, 15);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp1, 11);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLead(TP, 13);
+ metricsManager.recordPartitionLead(tp1, 13);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(11, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags), EPSILON);
- assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags1), EPSILON);
+ assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags1), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLead(tp2, 18);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp2, 12);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLead(tp2, 15);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(12, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, tags2), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, tags2), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, deprecatedTags), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, deprecatedTags), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, deprecatedTags), EPSILON);
+ }
+
+ @Test
+ public void testMaybeUpdateAssignment() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+ int initialMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1));
+
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // 1 new metrics shall be registered.
+ assertEquals(1, metrics.metrics().size() - initialMetricsSize);
+
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ subscriptionState.updatePreferredReadReplica(tp2, 1, () -> 0L);
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Another 2 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ // Validate preferred read replica metrics.
+ assertEquals(-1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags1), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags2), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, deprecatedTags), EPSILON);
+
+ // Remove tp2 from subscription set.
+ subscriptionState.assignFromUser(Set.of(tp1, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall remain same as tp2 should be removed and tp3 gets added.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ // Remove all partitions.
+ subscriptionState.assignFromUser(Set.of());
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall be same as initial count as all new metrics shall be removed.
+ assertEquals(initialMetricsSize, metrics.metrics().size());
+ }
+
+ @Test
+ public void testMaybeUpdateAssignmentWithAdditionalRegisteredMetrics() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ metricsManager.recordPartitionLead(tp1, 11);
+ metricsManager.recordPartitionLag(tp2, 5);
+ metricsManager.recordPartitionLead(tp2, 1);
+ metricsManager.recordPartitionLag(tp3, 4);
+ metricsManager.recordPartitionLead(tp3, 2);
+
+ int additionalRegisteredMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1, tp2, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+
+ // 5 new metrics shall be registered.
+ assertEquals(5, metrics.metrics().size() - additionalRegisteredMetricsSize);
+
+ // Remove 1 topic which has deprecated metrics as well.
Review Comment:
Done.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1894271882
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +198,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
+ if (shouldReportDeprecatedMetric(topic)) {
+ Sensor deprecatedBytesFetched = new SensorBuilder(metrics, deprecatedMetricName(name), () -> topicTags(topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
+ deprecatedBytesFetched.record(bytes);
+ }
+ }
+
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedRecordsFetched(String name, String topic, int records) {
+ if (shouldReportDeprecatedMetric(topic)) {
+ Sensor deprecatedRecordsFetched = new SensorBuilder(metrics, deprecatedMetricName(name), () -> topicTags(topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
+ deprecatedRecordsFetched.record(records);
+ }
+ }
+
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedPartitionLag(String name, TopicPartition tp, long lag) {
+ if (shouldReportDeprecatedMetric(tp.topic())) {
+ Sensor deprecatedRecordsLag = new SensorBuilder(metrics, deprecatedMetricName(name), () -> topicPartitionTags(tp))
+ .withValue(metricsRegistry.partitionRecordsLag)
+ .withMax(metricsRegistry.partitionRecordsLagMax)
+ .withAvg(metricsRegistry.partitionRecordsLagAvg)
+ .build();
+
+ deprecatedRecordsLag.record(lag);
+ }
+ }
+
+ @Deprecated // To be removed in Kafka 5.0 release.
Review Comment:
Could you please open the jira to ensure we will remove them from 5.0 :)
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1892613183
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +195,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
Review Comment:
As the metrics with a specfic tag has been deprectaed hence it's hard to specify same metric as deprecated. hence I have added `Note` to consumer topic metrics. For consumer metrics ops need not to ne updated as ops.html uses metrics description to generate details on kafka-site, I have verified that the added `Note` can be viewed on kafka-site locally.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #18232:
URL: https://github.com/apache/kafka/pull/18232#issuecomment-2556844587
Also, is the test failure related to this PR?
It seems to be build scan publish failure. I ll keep an eye on current ongoing build to see if it passes.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900780156
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +195,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
Review Comment:
Sure, so currently below `Note` has been added to metrics.
#### Current:
```
"Note: For topic names with periods (.), an additional metric with underscores is emitted.
However, the periods replaced metric is deprecated.
Please use the metric with actual topic name instead."
```
Example:
<img width="247" alt="Screenshot 2025-01-02 at 11 21 27" src="https://github.com/user-attachments/assets/f5395b1c-f6df-4006-b303-63b4db3c08d8" />
#### Should we change it to:
```
"Note: For topic names with periods (.), an additional metric with underscores is emitted.
However, the periods replaced metric is deprecated since Kafka 4.1 and shall be removed in Kafka 5.0.
Please use the metric with actual topic name instead."
```
wdyt?
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1889943665
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +101,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> Map.of("topic", tp.topic(), "partition", String.valueOf(tp.partition())))
Review Comment:
nit: Could you please consider placing "topic" before "partition"? See the following screenshot:
That can be addressed by using `Utils.mkMap`
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1892444217
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -145,46 +167,209 @@ public void testRecordsFetched() {
@Test
public void testRecordsFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordRecordsFetched(topicName1, 2);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordRecordsFetched(topicName2, 1);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for topicName2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
time.sleep(metrics.config().timeWindowMs() + 1);
metricsManager.recordRecordsFetched(topicName1, 10);
metricsManager.recordRecordsFetched(topicName2, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for topicName1.
assertEquals(6, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags1), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags1) > 0);
+ assertEquals(12, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags1), EPSILON);
+ // Validate metrics for topicName2.
assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags2), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags2) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, deprecatedTags), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, deprecatedTags) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLag() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLag(TP, 14);
- metricsManager.recordPartitionLag(TP, 8);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLag(tp1, 8);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLag(TP, 5);
+ metricsManager.recordPartitionLag(tp1, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(14, metricValue(metricsRegistry.recordsLagMax), EPSILON);
- assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags), EPSILON);
- assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags), EPSILON);
- assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags), EPSILON);
+ assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags1), EPSILON);
+ assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags1), EPSILON);
+ assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLag(tp2, 7);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ metricsManager.recordPartitionLag(tp2, 3);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLag(tp2, 2);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(7, metricValue(metricsRegistry.recordsLagMax), EPSILON);
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, tags2), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, tags2), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, deprecatedTags), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, deprecatedTags), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLead() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLead(TP, 15);
- metricsManager.recordPartitionLead(TP, 11);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLead(tp1, 15);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp1, 11);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLead(TP, 13);
+ metricsManager.recordPartitionLead(tp1, 13);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(11, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags), EPSILON);
- assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags1), EPSILON);
+ assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags1), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLead(tp2, 18);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp2, 12);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLead(tp2, 15);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(12, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, tags2), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, tags2), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, deprecatedTags), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, deprecatedTags), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, deprecatedTags), EPSILON);
+ }
+
+ @Test
+ public void testMaybeUpdateAssignment() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+ int initialMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1));
+
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // 1 new metrics shall be registered.
+ assertEquals(1, metrics.metrics().size() - initialMetricsSize);
+
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ subscriptionState.updatePreferredReadReplica(tp2, 1, () -> 0L);
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Another 2 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ // Validate preferred read replica metrics.
+ assertEquals(-1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags1), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags2), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, deprecatedTags), EPSILON);
+
+ // Remove tp2 from subscription set.
+ subscriptionState.assignFromUser(Set.of(tp1, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall remain same as tp2 should be removed and tp3 gets added.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ // Remove all partitions.
+ subscriptionState.assignFromUser(Set.of());
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall be same as initial count as all new metrics shall be removed.
+ assertEquals(initialMetricsSize, metrics.metrics().size());
+ }
+
+ @Test
+ public void testMaybeUpdateAssignmentWithAdditionalRegisteredMetrics() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ metricsManager.recordPartitionLead(tp1, 11);
+ metricsManager.recordPartitionLag(tp2, 5);
+ metricsManager.recordPartitionLead(tp2, 1);
+ metricsManager.recordPartitionLag(tp3, 4);
+ metricsManager.recordPartitionLead(tp3, 2);
+
+ int additionalRegisteredMetricsSize = metrics.metrics().size();
Review Comment:
I haven't asserted them on purpose as above tests already validated them.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1892615405
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -145,46 +167,209 @@ public void testRecordsFetched() {
@Test
public void testRecordsFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordRecordsFetched(topicName1, 2);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordRecordsFetched(topicName2, 1);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for topicName2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
time.sleep(metrics.config().timeWindowMs() + 1);
metricsManager.recordRecordsFetched(topicName1, 10);
metricsManager.recordRecordsFetched(topicName2, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for topicName1.
assertEquals(6, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags1), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags1) > 0);
+ assertEquals(12, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags1), EPSILON);
+ // Validate metrics for topicName2.
assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags2), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags2) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, deprecatedTags), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, deprecatedTags) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLag() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLag(TP, 14);
- metricsManager.recordPartitionLag(TP, 8);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLag(tp1, 8);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLag(TP, 5);
+ metricsManager.recordPartitionLag(tp1, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(14, metricValue(metricsRegistry.recordsLagMax), EPSILON);
- assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags), EPSILON);
- assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags), EPSILON);
- assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags), EPSILON);
+ assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags1), EPSILON);
+ assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags1), EPSILON);
+ assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLag(tp2, 7);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ metricsManager.recordPartitionLag(tp2, 3);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLag(tp2, 2);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(7, metricValue(metricsRegistry.recordsLagMax), EPSILON);
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, tags2), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, tags2), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, deprecatedTags), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, deprecatedTags), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLead() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLead(TP, 15);
- metricsManager.recordPartitionLead(TP, 11);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLead(tp1, 15);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp1, 11);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLead(TP, 13);
+ metricsManager.recordPartitionLead(tp1, 13);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(11, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags), EPSILON);
- assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags1), EPSILON);
+ assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags1), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLead(tp2, 18);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp2, 12);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLead(tp2, 15);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(12, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, tags2), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, tags2), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, deprecatedTags), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, deprecatedTags), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, deprecatedTags), EPSILON);
+ }
+
+ @Test
+ public void testMaybeUpdateAssignment() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+ int initialMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1));
+
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // 1 new metrics shall be registered.
+ assertEquals(1, metrics.metrics().size() - initialMetricsSize);
+
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ subscriptionState.updatePreferredReadReplica(tp2, 1, () -> 0L);
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Another 2 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ // Validate preferred read replica metrics.
+ assertEquals(-1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags1), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags2), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, deprecatedTags), EPSILON);
+
+ // Remove tp2 from subscription set.
+ subscriptionState.assignFromUser(Set.of(tp1, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall remain same as tp2 should be removed and tp3 gets added.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ // Remove all partitions.
+ subscriptionState.assignFromUser(Set.of());
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall be same as initial count as all new metrics shall be removed.
+ assertEquals(initialMetricsSize, metrics.metrics().size());
+ }
+
+ @Test
+ public void testMaybeUpdateAssignmentWithAdditionalRegisteredMetrics() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ metricsManager.recordPartitionLead(tp1, 11);
+ metricsManager.recordPartitionLag(tp2, 5);
+ metricsManager.recordPartitionLead(tp2, 1);
+ metricsManager.recordPartitionLag(tp3, 4);
+ metricsManager.recordPartitionLead(tp3, 2);
+
+ int additionalRegisteredMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1, tp2, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+
+ // 5 new metrics shall be registered.
+ assertEquals(5, metrics.metrics().size() - additionalRegisteredMetricsSize);
+
+ // Remove 1 topic which has deprecated metrics as well.
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // For tp2, 14 metrics will be unregistered. 3 for partition lag, 3 for partition lead, 1 for
+ // preferred read replica and similarly 7 deprecated metrics.
+ assertEquals(-9, metrics.metrics().size() - additionalRegisteredMetricsSize);
Review Comment:
I have added a comment and toggled the check, please let me know if it's better now.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -32,18 +33,19 @@
import org.junit.jupiter.api.Test;
import java.util.Map;
+import java.util.Set;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicPartitionTags;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicTags;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class FetchMetricsManagerTest {
Review Comment:
Done.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1892715492
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +101,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> Map.of("topic", tp.topic(), "partition", String.valueOf(tp.partition())))
Review Comment:
Make sense, I earlier considered it but didn't do as previous tags were not ordered. I have added the change.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1898435818
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -24,15 +24,18 @@
import org.apache.kafka.common.metrics.stats.WindowedCount;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+
/**
* The {@link FetchMetricsManager} class provides wrapper methods to record lag, lead, latency, and fetch metrics.
* It keeps an internal ID of the assigned set of partitions which is updated to ensure the set of metrics it
* records matches up with the topic-partitions in use.
*/
+@SuppressWarnings("deprecation")
Review Comment:
I don't believe this suppression has any value because it's implementing deprecated code, as opposed to using it.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -32,18 +33,19 @@
import org.junit.jupiter.api.Test;
import java.util.Map;
+import java.util.Set;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicPartitionTags;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicTags;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class FetchMetricsManagerTest {
Review Comment:
I have a suspicion you put the suppression on `FetchMetricsManager` when I meant to put it on this test.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900780156
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +195,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
Review Comment:
Sure, so currently below `Note` has been added to metrics.
#### Current:
```
"Note: For topic names with periods (.), an additional metric with underscores is emitted.
However, the periods replaced metric is deprecated.
Please use the metric with actual topic name instead."
```
Example:
<img width="247" alt="Screenshot 2025-01-02 at 11 21 27" src="https://github.com/user-attachments/assets/f5395b1c-f6df-4006-b303-63b4db3c08d8" />
#### Should we change it to:
```
"Note: For topic names with periods (.), an additional metric with underscores is emitted.
However, the periods replaced metric is deprecated since Kafka 4.1 and shall be removed in Kafka 5.0.
Please use the metric with actual topic name instead."
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900781235
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -197,22 +274,34 @@ private static String partitionRecordsLagMetricName(TopicPartition tp) {
return tp + ".records-lag";
}
+ private static String deprecatedMetricName(String name) {
+ return name + ".deprecated";
+ }
+
+ private static boolean shouldReportDeprecatedMetric(String topic) {
+ return topic.contains(".");
+ }
+
private MetricName partitionPreferredReadReplicaMetricName(TopicPartition tp) {
+ Map<String, String> metricTags = mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition())));
+ return this.metrics.metricInstance(metricsRegistry.partitionPreferredReadReplica, metricTags);
+ }
+
+ @Deprecated
Review Comment:
Sure, I have added details here https://github.com/apache/kafka/pull/18232#discussion_r1900780156, please let me know your thoughts.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1892613183
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +195,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
Review Comment:
As the metrics with a specfic tag has been deprectaed hence it's hard to specify same metric as deprecated. Hence I have added `Note` to consumer topic metrics. For consumer metrics ops need not to ne updated as ops.html uses metrics description to generate details on kafka-site, I have verified that the added `Note` can be viewed on kafka-site locally. Please let me know if you think it should be handled differently.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1893842482
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
Done.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900223373
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
Is there a reason why you're still using the `Utils.mkMap` methods instead of `Map.of` from Java 9+?
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900782037
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
I have opened https://issues.apache.org/jira/browse/KAFKA-18390 to address @ijuma's comment. The fix is targeted for version 4.1.0, as it may introduce significant changes to the codebase.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900424725
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -197,22 +274,34 @@ private static String partitionRecordsLagMetricName(TopicPartition tp) {
return tp + ".records-lag";
}
+ private static String deprecatedMetricName(String name) {
+ return name + ".deprecated";
+ }
+
+ private static boolean shouldReportDeprecatedMetric(String topic) {
+ return topic.contains(".");
+ }
+
private MetricName partitionPreferredReadReplicaMetricName(TopicPartition tp) {
+ Map<String, String> metricTags = mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition())));
+ return this.metrics.metricInstance(metricsRegistry.partitionPreferredReadReplica, metricTags);
+ }
+
+ @Deprecated
Review Comment:
We should never deprecate something without including a @deprecated javadoc with the details I outlined in a separate comment.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1893012622
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
Could we do the same for the deprecated metrics too for better consistency?
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1888991498
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -162,16 +170,22 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
metrics.removeSensor(partitionRecordsLagMetricName(tp));
metrics.removeSensor(partitionRecordsLeadMetricName(tp));
metrics.removeMetric(partitionPreferredReadReplicaMetricName(tp));
+ // Remove deprecated metrics.
+ metrics.removeSensor(deprecatedMetricName(partitionRecordsLagMetricName(tp)));
+ metrics.removeSensor(deprecatedMetricName(partitionRecordsLeadMetricName(tp)));
+ metrics.removeMetric(partitionPreferredReadReplicaMetricNameDeprecated(tp));
Review Comment:
partitionPreferredReadReplicaMetricNameDeprecated => deprecatedPartitionPreferredReadReplicaMetricName to be consistent with other names?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -117,20 +119,40 @@ public void testBytesFetched() {
@Test
public void testBytesFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordBytesFetched(topicName1, 2);
+ // 4 new metrics shall be registered.
+ assertEquals(4, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordBytesFetched(topicName2, 1);
+ // Another 8 metrics gets registered as deprecated metrics should be reported for topicName2.
Review Comment:
gets => get
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +195,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
Review Comment:
It would be useful to follow the approach used in https://github.com/apache/kafka/pull/11302/files to add "deprecated" in the description of the metric name and update the ops doc about the deprecation.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -145,46 +167,209 @@ public void testRecordsFetched() {
@Test
public void testRecordsFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordRecordsFetched(topicName1, 2);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordRecordsFetched(topicName2, 1);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for topicName2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
time.sleep(metrics.config().timeWindowMs() + 1);
metricsManager.recordRecordsFetched(topicName1, 10);
metricsManager.recordRecordsFetched(topicName2, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for topicName1.
assertEquals(6, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags1), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags1) > 0);
+ assertEquals(12, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags1), EPSILON);
+ // Validate metrics for topicName2.
assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags2), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags2) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, deprecatedTags), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, deprecatedTags) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLag() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLag(TP, 14);
- metricsManager.recordPartitionLag(TP, 8);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLag(tp1, 8);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLag(TP, 5);
+ metricsManager.recordPartitionLag(tp1, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(14, metricValue(metricsRegistry.recordsLagMax), EPSILON);
- assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags), EPSILON);
- assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags), EPSILON);
- assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags), EPSILON);
+ assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags1), EPSILON);
+ assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags1), EPSILON);
+ assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLag(tp2, 7);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ metricsManager.recordPartitionLag(tp2, 3);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLag(tp2, 2);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(7, metricValue(metricsRegistry.recordsLagMax), EPSILON);
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, tags2), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, tags2), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, deprecatedTags), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, deprecatedTags), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLead() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLead(TP, 15);
- metricsManager.recordPartitionLead(TP, 11);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLead(tp1, 15);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp1, 11);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLead(TP, 13);
+ metricsManager.recordPartitionLead(tp1, 13);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(11, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags), EPSILON);
- assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags1), EPSILON);
+ assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags1), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLead(tp2, 18);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp2, 12);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLead(tp2, 15);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(12, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, tags2), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, tags2), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, deprecatedTags), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, deprecatedTags), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, deprecatedTags), EPSILON);
+ }
+
+ @Test
+ public void testMaybeUpdateAssignment() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+ int initialMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1));
+
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // 1 new metrics shall be registered.
+ assertEquals(1, metrics.metrics().size() - initialMetricsSize);
+
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ subscriptionState.updatePreferredReadReplica(tp2, 1, () -> 0L);
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Another 2 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ // Validate preferred read replica metrics.
+ assertEquals(-1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags1), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags2), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, deprecatedTags), EPSILON);
+
+ // Remove tp2 from subscription set.
+ subscriptionState.assignFromUser(Set.of(tp1, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall remain same as tp2 should be removed and tp3 gets added.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ // Remove all partitions.
+ subscriptionState.assignFromUser(Set.of());
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall be same as initial count as all new metrics shall be removed.
+ assertEquals(initialMetricsSize, metrics.metrics().size());
+ }
+
+ @Test
+ public void testMaybeUpdateAssignmentWithAdditionalRegisteredMetrics() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ metricsManager.recordPartitionLead(tp1, 11);
+ metricsManager.recordPartitionLag(tp2, 5);
+ metricsManager.recordPartitionLead(tp2, 1);
+ metricsManager.recordPartitionLag(tp3, 4);
+ metricsManager.recordPartitionLead(tp3, 2);
+
+ int additionalRegisteredMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1, tp2, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+
+ // 5 new metrics shall be registered.
+ assertEquals(5, metrics.metrics().size() - additionalRegisteredMetricsSize);
+
+ // Remove 1 topic which has deprecated metrics as well.
Review Comment:
1 topic => 1 partition
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #18232:
URL: https://github.com/apache/kafka/pull/18232#issuecomment-2559541357
@AndrewJSchofield Please let us know if it's good to merge?
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1899487124
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -24,15 +24,18 @@
import org.apache.kafka.common.metrics.stats.WindowedCount;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+
/**
* The {@link FetchMetricsManager} class provides wrapper methods to record lag, lead, latency, and fetch metrics.
* It keeps an internal ID of the assigned set of partitions which is updated to ensure the set of metrics it
* records matches up with the topic-partitions in use.
*/
+@SuppressWarnings("deprecation")
Review Comment:
Yes you are right, I have corrected it.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1894344240
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -181,6 +198,67 @@ void maybeUpdateAssignment(SubscriptionState subscription) {
}
}
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedBytesFetched(String name, String topic, int bytes) {
+ if (shouldReportDeprecatedMetric(topic)) {
+ Sensor deprecatedBytesFetched = new SensorBuilder(metrics, deprecatedMetricName(name), () -> topicTags(topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
+ deprecatedBytesFetched.record(bytes);
+ }
+ }
+
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedRecordsFetched(String name, String topic, int records) {
+ if (shouldReportDeprecatedMetric(topic)) {
+ Sensor deprecatedRecordsFetched = new SensorBuilder(metrics, deprecatedMetricName(name), () -> topicTags(topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
+ deprecatedRecordsFetched.record(records);
+ }
+ }
+
+ @Deprecated // To be removed in Kafka 5.0 release.
+ private void maybeRecordDeprecatedPartitionLag(String name, TopicPartition tp, long lag) {
+ if (shouldReportDeprecatedMetric(tp.topic())) {
+ Sensor deprecatedRecordsLag = new SensorBuilder(metrics, deprecatedMetricName(name), () -> topicPartitionTags(tp))
+ .withValue(metricsRegistry.partitionRecordsLag)
+ .withMax(metricsRegistry.partitionRecordsLagMax)
+ .withAvg(metricsRegistry.partitionRecordsLagAvg)
+ .build();
+
+ deprecatedRecordsLag.record(lag);
+ }
+ }
+
+ @Deprecated // To be removed in Kafka 5.0 release.
Review Comment:
Done: https://issues.apache.org/jira/browse/KAFKA-18335
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1900425014
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java:
##########
@@ -101,32 +105,38 @@ void recordRecordsFetched(int records) {
void recordBytesFetched(String topic, int bytes) {
String name = topicBytesFetchedMetricName(topic);
- Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicFetchSizeAvg)
- .withMax(metricsRegistry.topicFetchSizeMax)
- .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
- .build();
+ maybeRecordDeprecatedBytesFetched(name, topic, bytes);
+
+ Sensor bytesFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicFetchSizeAvg)
+ .withMax(metricsRegistry.topicFetchSizeMax)
+ .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
+ .build();
bytesFetched.record(bytes);
}
void recordRecordsFetched(String topic, int records) {
String name = topicRecordsFetchedMetricName(topic);
- Sensor recordsFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
- .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
- .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
- .build();
+ maybeRecordDeprecatedRecordsFetched(name, topic, records);
+
+ Sensor recordsFetched = new SensorBuilder(metrics, name, () -> Map.of("topic", topic))
+ .withAvg(metricsRegistry.topicRecordsPerRequestAvg)
+ .withMeter(metricsRegistry.topicRecordsConsumedRate, metricsRegistry.topicRecordsConsumedTotal)
+ .build();
recordsFetched.record(records);
}
void recordPartitionLag(TopicPartition tp, long lag) {
this.recordsLag.record(lag);
String name = partitionRecordsLagMetricName(tp);
- Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
- .withValue(metricsRegistry.partitionRecordsLag)
- .withMax(metricsRegistry.partitionRecordsLagMax)
- .withAvg(metricsRegistry.partitionRecordsLagAvg)
- .build();
+ maybeRecordDeprecatedPartitionLag(name, tp, lag);
+
+ Sensor recordsLag = new SensorBuilder(metrics, name, () -> mkMap(mkEntry("topic", tp.topic()), mkEntry("partition", String.valueOf(tp.partition()))))
Review Comment:
Hmm, if we want ordering to be preserved, we should change `SensorBuilder` to use a `LinkedHashMap` (`SequencedMap` would be better, but that requires Java 21). Having this implicit requirement and hoping people get it right is very error prone.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1899487576
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -32,18 +33,19 @@
import org.junit.jupiter.api.Test;
import java.util.Map;
+import java.util.Set;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicPartitionTags;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicTags;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class FetchMetricsManagerTest {
Review Comment:
I have corrected it, also I have placed the suppression on individual tests so they can be removed while removing deprecated code.
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #18232:
URL: https://github.com/apache/kafka/pull/18232
--
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
Re: [PR] KAFKA-12469: Deprecated and corrected topic metrics for consumer (KIP-1109) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18232:
URL: https://github.com/apache/kafka/pull/18232#discussion_r1890416696
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -145,46 +167,209 @@ public void testRecordsFetched() {
@Test
public void testRecordsFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordRecordsFetched(topicName1, 2);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordRecordsFetched(topicName2, 1);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for topicName2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
time.sleep(metrics.config().timeWindowMs() + 1);
metricsManager.recordRecordsFetched(topicName1, 10);
metricsManager.recordRecordsFetched(topicName2, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for topicName1.
assertEquals(6, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags1), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags1) > 0);
+ assertEquals(12, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags1), EPSILON);
+ // Validate metrics for topicName2.
assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags2), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags2) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, deprecatedTags), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, deprecatedTags) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLag() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLag(TP, 14);
- metricsManager.recordPartitionLag(TP, 8);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLag(tp1, 8);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLag(TP, 5);
+ metricsManager.recordPartitionLag(tp1, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(14, metricValue(metricsRegistry.recordsLagMax), EPSILON);
- assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags), EPSILON);
- assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags), EPSILON);
- assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags), EPSILON);
+ assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags1), EPSILON);
+ assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags1), EPSILON);
+ assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLag(tp2, 7);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ metricsManager.recordPartitionLag(tp2, 3);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLag(tp2, 2);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(7, metricValue(metricsRegistry.recordsLagMax), EPSILON);
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, tags2), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, tags2), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, deprecatedTags), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, deprecatedTags), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLead() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLead(TP, 15);
- metricsManager.recordPartitionLead(TP, 11);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLead(tp1, 15);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp1, 11);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLead(TP, 13);
+ metricsManager.recordPartitionLead(tp1, 13);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(11, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags), EPSILON);
- assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags1), EPSILON);
+ assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags1), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLead(tp2, 18);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp2, 12);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLead(tp2, 15);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(12, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, tags2), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, tags2), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, deprecatedTags), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, deprecatedTags), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, deprecatedTags), EPSILON);
+ }
+
+ @Test
+ public void testMaybeUpdateAssignment() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+ int initialMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1));
+
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // 1 new metrics shall be registered.
+ assertEquals(1, metrics.metrics().size() - initialMetricsSize);
+
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ subscriptionState.updatePreferredReadReplica(tp2, 1, () -> 0L);
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Another 2 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ // Validate preferred read replica metrics.
+ assertEquals(-1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags1), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags2), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, deprecatedTags), EPSILON);
+
+ // Remove tp2 from subscription set.
+ subscriptionState.assignFromUser(Set.of(tp1, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall remain same as tp2 should be removed and tp3 gets added.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ // Remove all partitions.
+ subscriptionState.assignFromUser(Set.of());
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall be same as initial count as all new metrics shall be removed.
+ assertEquals(initialMetricsSize, metrics.metrics().size());
+ }
+
+ @Test
+ public void testMaybeUpdateAssignmentWithAdditionalRegisteredMetrics() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ metricsManager.recordPartitionLead(tp1, 11);
+ metricsManager.recordPartitionLag(tp2, 5);
+ metricsManager.recordPartitionLead(tp2, 1);
+ metricsManager.recordPartitionLag(tp3, 4);
+ metricsManager.recordPartitionLead(tp3, 2);
+
+ int additionalRegisteredMetricsSize = metrics.metrics().size();
Review Comment:
Wouldn't it be sensible to assert the number of additional registered metrics too?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -32,18 +33,19 @@
import org.junit.jupiter.api.Test;
import java.util.Map;
+import java.util.Set;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicPartitionTags;
import static org.apache.kafka.clients.consumer.internals.FetchMetricsManager.topicTags;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class FetchMetricsManagerTest {
Review Comment:
Personally, I would put `@SuppressWarnings("deprecation")` on this class because it is knowingly using deprecated methods.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManagerTest.java:
##########
@@ -145,46 +167,209 @@ public void testRecordsFetched() {
@Test
public void testRecordsFetchedTopic() {
String topicName1 = TOPIC_NAME;
- String topicName2 = "another-topic";
- Map<String, String> tags1 = topicTags(topicName1);
- Map<String, String> tags2 = topicTags(topicName2);
+ String topicName2 = "another.topic";
+ Map<String, String> tags1 = Map.of("topic", topicName1);
+ Map<String, String> tags2 = Map.of("topic", topicName2);
+ Map<String, String> deprecatedTags = topicTags(topicName2);
+ int initialMetricsSize = metrics.metrics().size();
metricsManager.recordRecordsFetched(topicName1, 2);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
metricsManager.recordRecordsFetched(topicName2, 1);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for topicName2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
time.sleep(metrics.config().timeWindowMs() + 1);
metricsManager.recordRecordsFetched(topicName1, 10);
metricsManager.recordRecordsFetched(topicName2, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for topicName1.
assertEquals(6, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags1), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags1) > 0);
+ assertEquals(12, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags1), EPSILON);
+ // Validate metrics for topicName2.
assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, tags2), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, tags2) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(3, metricValue(metricsRegistry.topicRecordsPerRequestAvg, deprecatedTags), EPSILON);
+ assertTrue(metricValue(metricsRegistry.topicRecordsConsumedRate, deprecatedTags) > 0);
+ assertEquals(6, metricValue(metricsRegistry.topicRecordsConsumedTotal, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLag() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLag(TP, 14);
- metricsManager.recordPartitionLag(TP, 8);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLag(tp1, 8);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLag(TP, 5);
+ metricsManager.recordPartitionLag(tp1, 5);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(14, metricValue(metricsRegistry.recordsLagMax), EPSILON);
- assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags), EPSILON);
- assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags), EPSILON);
- assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags), EPSILON);
+ assertEquals(5, metricValue(metricsRegistry.partitionRecordsLag, tags1), EPSILON);
+ assertEquals(14, metricValue(metricsRegistry.partitionRecordsLagMax, tags1), EPSILON);
+ assertEquals(9, metricValue(metricsRegistry.partitionRecordsLagAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLag(tp2, 7);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ metricsManager.recordPartitionLag(tp2, 3);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLag(tp2, 2);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(7, metricValue(metricsRegistry.recordsLagMax), EPSILON);
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, tags2), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, tags2), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(2, metricValue(metricsRegistry.partitionRecordsLag, deprecatedTags), EPSILON);
+ assertEquals(7, metricValue(metricsRegistry.partitionRecordsLagMax, deprecatedTags), EPSILON);
+ assertEquals(4, metricValue(metricsRegistry.partitionRecordsLagAvg, deprecatedTags), EPSILON);
}
@Test
public void testPartitionLead() {
- Map<String, String> tags = topicPartitionTags(TP);
- metricsManager.recordPartitionLead(TP, 15);
- metricsManager.recordPartitionLead(TP, 11);
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLead(tp1, 15);
+ // 3 new metrics shall be registered.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp1, 11);
time.sleep(metrics.config().timeWindowMs() + 1);
- metricsManager.recordPartitionLead(TP, 13);
+ metricsManager.recordPartitionLead(tp1, 13);
+ // Subsequent calls should not register new metrics.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp1.
assertEquals(11, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags), EPSILON);
- assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags), EPSILON);
- assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLead, tags1), EPSILON);
+ assertEquals(11, metricValue(metricsRegistry.partitionRecordsLeadMin, tags1), EPSILON);
+ assertEquals(13, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags1), EPSILON);
+
+ metricsManager.recordPartitionLead(tp2, 18);
+ // Another 6 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+
+ metricsManager.recordPartitionLead(tp2, 12);
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ metricsManager.recordPartitionLead(tp2, 15);
+
+ // Subsequent calls should not register new metrics.
+ assertEquals(9, metrics.metrics().size() - initialMetricsSize);
+ // Validate metrics for tp2.
+ assertEquals(12, metricValue(metricsRegistry.recordsLeadMin), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, tags2), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, tags2), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, tags2), EPSILON);
+ // Validate metrics for deprecated topic.
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLead, deprecatedTags), EPSILON);
+ assertEquals(12, metricValue(metricsRegistry.partitionRecordsLeadMin, deprecatedTags), EPSILON);
+ assertEquals(15, metricValue(metricsRegistry.partitionRecordsLeadAvg, deprecatedTags), EPSILON);
+ }
+
+ @Test
+ public void testMaybeUpdateAssignment() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+ int initialMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1));
+
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // 1 new metrics shall be registered.
+ assertEquals(1, metrics.metrics().size() - initialMetricsSize);
+
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ subscriptionState.updatePreferredReadReplica(tp2, 1, () -> 0L);
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Another 2 metrics gets registered as deprecated metrics should be reported for tp2.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ Map<String, String> tags1 = Map.of("topic", tp1.topic(), "partition", String.valueOf(tp1.partition()));
+ Map<String, String> tags2 = Map.of("topic", tp2.topic(), "partition", String.valueOf(tp2.partition()));
+ Map<String, String> deprecatedTags = topicPartitionTags(tp2);
+ // Validate preferred read replica metrics.
+ assertEquals(-1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags1), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, tags2), EPSILON);
+ assertEquals(1, readReplicaMetricValue(metricsRegistry.partitionPreferredReadReplica, deprecatedTags), EPSILON);
+
+ // Remove tp2 from subscription set.
+ subscriptionState.assignFromUser(Set.of(tp1, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall remain same as tp2 should be removed and tp3 gets added.
+ assertEquals(3, metrics.metrics().size() - initialMetricsSize);
+
+ // Remove all partitions.
+ subscriptionState.assignFromUser(Set.of());
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // Metrics count shall be same as initial count as all new metrics shall be removed.
+ assertEquals(initialMetricsSize, metrics.metrics().size());
+ }
+
+ @Test
+ public void testMaybeUpdateAssignmentWithAdditionalRegisteredMetrics() {
+ TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 0);
+ TopicPartition tp2 = new TopicPartition("another.topic", 0);
+ TopicPartition tp3 = new TopicPartition("another.topic", 1);
+
+ int initialMetricsSize = metrics.metrics().size();
+
+ metricsManager.recordPartitionLag(tp1, 14);
+ metricsManager.recordPartitionLead(tp1, 11);
+ metricsManager.recordPartitionLag(tp2, 5);
+ metricsManager.recordPartitionLead(tp2, 1);
+ metricsManager.recordPartitionLag(tp3, 4);
+ metricsManager.recordPartitionLead(tp3, 2);
+
+ int additionalRegisteredMetricsSize = metrics.metrics().size();
+
+ SubscriptionState subscriptionState = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+ subscriptionState.assignFromUser(Set.of(tp1, tp2, tp3));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+
+ // 5 new metrics shall be registered.
+ assertEquals(5, metrics.metrics().size() - additionalRegisteredMetricsSize);
+
+ // Remove 1 topic which has deprecated metrics as well.
+ subscriptionState.assignFromUser(Set.of(tp1, tp2));
+ metricsManager.maybeUpdateAssignment(subscriptionState);
+ // For tp2, 14 metrics will be unregistered. 3 for partition lag, 3 for partition lead, 1 for
+ // preferred read replica and similarly 7 deprecated metrics.
+ assertEquals(-9, metrics.metrics().size() - additionalRegisteredMetricsSize);
Review Comment:
This -9 is needlessly obscure.
--
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