Skip to content

KAFKA-5890 records.lag should use tags for topic and partition rather than using metric name. #4362

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
lahabana wants to merge 1 commit into from
Closed

KAFKA-5890 records.lag should use tags for topic and partition rather than using metric name. #4362

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1327,16 +1327,25 @@ private void recordPartitionLag(TopicPartition tp, long lag) {
String name = partitionLagMetricName(tp);
Sensor recordsLag = this.metrics.getSensor(name);
if (recordsLag == null) {
Map<String, String> metricTags = new HashMap<>(2);
metricTags.put("topic", tp.topic().replace('.', '_'));

@makubi makubi Jul 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Is there a reason you replace '.' by '_' in that case?

@lahabana lahabana Jul 19, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

this is always done in metrics. I believe it's because JMX uses . to define hierarchy on entities so using a dot here would cause issues in the jmx reporter.
It's mentioned here: http://kafka.apache.org/documentation.html#upgrade_9_breaking

@makubi makubi Jul 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Thanks for your reply. I guess this may be true in some cases, but at least for e.g. kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec it works even if the topic contains dots.

metricTags.put("partition", String.valueOf(tp.partition()));

recordsLag = this.metrics.sensor(name);

recordsLag.add(this.metrics.metricInstance(metricsRegistry.partitionRecordsLag, metricTags), new Value());
recordsLag.add(this.metrics.metricInstance(metricsRegistry.partitionRecordsLagMax, metricTags), new Max());
recordsLag.add(this.metrics.metricInstance(metricsRegistry.partitionRecordsLagAvg, metricTags), new Avg());

recordsLag.add(this.metrics.metricName(name,
metricsRegistry.partitionRecordsLag.group(),
metricsRegistry.partitionRecordsLag.description()), new Value());
metricsRegistry.partitionRecordsLagDeprecated.group(),
metricsRegistry.partitionRecordsLagDeprecated.description()), new Value());
recordsLag.add(this.metrics.metricName(name + "-max",
metricsRegistry.partitionRecordsLagMax.group(),
metricsRegistry.partitionRecordsLagMax.description()), new Max());
metricsRegistry.partitionRecordsLagMaxDeprecated.group(),
metricsRegistry.partitionRecordsLagMaxDeprecated.description()), new Max());
recordsLag.add(this.metrics.metricName(name + "-avg",
metricsRegistry.partitionRecordsLagAvg.group(),
metricsRegistry.partitionRecordsLagAvg.description()), new Avg());
metricsRegistry.partitionRecordsLagAvgDeprecated.group(),
metricsRegistry.partitionRecordsLagAvgDeprecated.description()), new Avg());
}
recordsLag.record(lag);
}
Expand Down
28 changes: 22 additions & 6 deletions ...nts/src/main/java/org/apache/kafka/clients/consumer/internals/FetcherMetricsRegistry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,10 @@ public class FetcherMetricsRegistry {
public MetricNameTemplate partitionRecordsLag;
public MetricNameTemplate partitionRecordsLagMax;
public MetricNameTemplate partitionRecordsLagAvg;
// To remove in 2.0
public MetricNameTemplate partitionRecordsLagDeprecated;
public MetricNameTemplate partitionRecordsLagMaxDeprecated;
public MetricNameTemplate partitionRecordsLagAvgDeprecated;

public FetcherMetricsRegistry() {
this(new HashSet<String>(), "");
Expand Down Expand Up @@ -117,13 +121,22 @@ public FetcherMetricsRegistry(Set<String> tags, String metricGrpPrefix) {
this.topicRecordsConsumedTotal = new MetricNameTemplate("records-consumed-total", groupName,
"The total number of records consumed for a topic", topicTags);

this.partitionRecordsLagDeprecated = new MetricNameTemplate("{topic}-{partition}.records-lag", groupName,
"The latest lag of the partition (DEPRECATED use the tag based version instead)", tags);
this.partitionRecordsLagMaxDeprecated = new MetricNameTemplate("{topic}-{partition}.records-lag-max", groupName,
"The max lag of the partition (DEPRECATED use the tag based version instead)", tags);
this.partitionRecordsLagAvgDeprecated = new MetricNameTemplate("{topic}-{partition}.records-lag-avg", groupName,
"The average lag of the partition (DEPRECATED use the tag based version instead)", tags);

/***** Partition level *****/
this.partitionRecordsLag = new MetricNameTemplate("{topic}-{partition}.records-lag", groupName,
"The latest lag of the partition", tags);
this.partitionRecordsLagMax = new MetricNameTemplate("{topic}-{partition}.records-lag-max", groupName,
"The max lag of the partition", tags);
this.partitionRecordsLagAvg = new MetricNameTemplate("{topic}-{partition}.records-lag-avg", groupName,
"The average lag of the partition", tags);
Set<String> partitionTags = new HashSet<>(topicTags);
partitionTags.add("partition");
this.partitionRecordsLag = new MetricNameTemplate("records-lag", groupName,
"The latest lag of the partition", partitionTags);
this.partitionRecordsLagMax = new MetricNameTemplate("records-lag-max", groupName,
"The max lag of the partition", partitionTags);
this.partitionRecordsLagAvg = new MetricNameTemplate("records-lag-avg", groupName,
"The average lag of the partition", partitionTags);


}
Expand Down Expand Up @@ -151,6 +164,9 @@ public List<MetricNameTemplate> getAllTemplates() {
topicRecordsPerRequestAvg,
topicRecordsConsumedRate,
topicRecordsConsumedTotal,
partitionRecordsLagDeprecated,
partitionRecordsLagAvgDeprecated,
partitionRecordsLagMaxDeprecated,
partitionRecordsLag,
partitionRecordsLagAvg,
partitionRecordsLagMax
Expand Down
16 changes: 14 additions & 2 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1171,7 +1171,10 @@ public void testFetcherMetrics() {
subscriptions.seek(tp0, 0);

MetricName maxLagMetric = metrics.metricInstance(metricsRegistry.recordsLagMax);
MetricName partitionLagMetric = metrics.metricName(tp0 + ".records-lag", metricGroup);
Map<String, String> tags = new HashMap<>();
tags.put("topic", tp0.topic());
tags.put("partition", String.valueOf(tp0.partition()));
MetricName partitionLagMetric = metrics.metricName("records-lag", metricGroup, tags);

Map<MetricName, KafkaMetric> allMetrics = metrics.metrics();
KafkaMetric recordsFetchLagMax = allMetrics.get(maxLagMetric);
Expand Down Expand Up @@ -1210,7 +1213,12 @@ public void testReadCommittedLagMetric() {
subscriptions.seek(tp0, 0);

MetricName maxLagMetric = metrics.metricInstance(metricsRegistry.recordsLagMax);
MetricName partitionLagMetric = metrics.metricName(tp0 + ".records-lag", metricGroup);

Map<String, String> tags = new HashMap<>();
tags.put("topic", tp0.topic());
tags.put("partition", String.valueOf(tp0.partition()));
MetricName partitionLagMetric = metrics.metricName("records-lag", metricGroup, tags);
MetricName partitionLagMetricDeprecated = metrics.metricName(tp0 + ".records-lag", metricGroup);

Map<MetricName, KafkaMetric> allMetrics = metrics.metrics();
KafkaMetric recordsFetchLagMax = allMetrics.get(maxLagMetric);
Expand All @@ -1225,6 +1233,9 @@ public void testReadCommittedLagMetric() {
KafkaMetric partitionLag = allMetrics.get(partitionLagMetric);
assertEquals(50, partitionLag.value(), EPSILON);

KafkaMetric partitionLagDeprecated = allMetrics.get(partitionLagMetricDeprecated);
assertEquals(50, partitionLagDeprecated.value(), EPSILON);

// recordsFetchLagMax should be lso - offset of the last message after receiving a non-empty FetchResponse
MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1024), CompressionType.NONE,
TimestampType.CREATE_TIME, 0L);
Expand All @@ -1237,6 +1248,7 @@ public void testReadCommittedLagMetric() {
// verify de-registration of partition lag
subscriptions.unsubscribe();
assertFalse(allMetrics.containsKey(partitionLagMetric));
assertFalse(allMetrics.containsKey(partitionLagMetricDeprecated));
}

@Test
Expand Down
41 changes: 33 additions & 8 deletions core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1398,8 +1398,16 @@ class PlaintextConsumerTest extends BaseConsumerTest {
}, "Consumer did not consume any message before timeout.")
assertEquals("should be assigned once", 1, listener0.callsToAssigned)
// Verify the metric exist.
val tags = Collections.singletonMap("client-id", "testPerPartitionLagMetricsCleanUpWithSubscribe")
val fetchLag0 = consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", tags))
val tags1 = new util.HashMap[String, String]()
tags1.put("client-id", "testPerPartitionLagMetricsCleanUpWithSubscribe")
tags1.put("topic", tp.topic())
tags1.put("partition", String.valueOf(tp.partition()))

val tags2 = new util.HashMap[String, String]()
tags2.put("client-id", "testPerPartitionLagMetricsCleanUpWithSubscribe")
tags2.put("topic", tp2.topic())
tags2.put("partition", String.valueOf(tp2.partition()))
val fetchLag0 = consumer.metrics.get(new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags1))
assertNotNull(fetchLag0)
val expectedLag = numMessages - records.count
assertEquals(s"The lag should be $expectedLag", expectedLag, fetchLag0.value, epsilon)
Expand All @@ -1411,8 +1419,8 @@ class PlaintextConsumerTest extends BaseConsumerTest {
listener0.callsToAssigned >= 2
}, "Expected rebalance did not occur.")
// Verify the metric has gone
assertNull(consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", tags)))
assertNull(consumer.metrics.get(new MetricName(tp2 + ".records-lag", "consumer-fetch-manager-metrics", "", tags)))
assertNull(consumer.metrics.get(new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags1)))
assertNull(consumer.metrics.get(new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags2)))
} finally {
consumer.close()
}
Expand All @@ -1436,15 +1444,24 @@ class PlaintextConsumerTest extends BaseConsumerTest {
!records.records(tp).isEmpty
}, "Consumer did not consume any message before timeout.")
// Verify the metric exist.
val tags = Collections.singletonMap("client-id", "testPerPartitionLagMetricsCleanUpWithAssign")
val fetchLag = consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", tags))
val tags = new util.HashMap[String, String]()
tags.put("client-id", "testPerPartitionLagMetricsCleanUpWithAssign")
tags.put("topic", tp.topic())
tags.put("partition", String.valueOf(tp.partition()))
val fetchLag = consumer.metrics.get(new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags))
assertNotNull(fetchLag)

val oldTags = Collections.singletonMap("client-id", "testPerPartitionLagMetricsCleanUpWithAssign")
val oldFetchLag = consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", oldTags))
assertEquals(fetchLag.metricValue(), oldFetchLag.metricValue())

val expectedLag = numMessages - records.count
assertEquals(s"The lag should be $expectedLag", expectedLag, fetchLag.value, epsilon)

consumer.assign(List(tp2).asJava)
TestUtils.waitUntilTrue(() => !consumer.poll(100).isEmpty, "Consumer did not consume any message before timeout.")
assertNull(consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", tags)))
assertNull(consumer.metrics.get(new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags)))
} finally {
consumer.close()
}
Expand All @@ -1466,8 +1483,16 @@ class PlaintextConsumerTest extends BaseConsumerTest {
records = consumer.poll(100)
!records.isEmpty
}, "Consumer did not consume any message before timeout.")
val tags = Collections.singletonMap("client-id", "testPerPartitionLagWithMaxPollRecords")
val lag = consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", tags))
val oldTags = Collections.singletonMap("client-id", "testPerPartitionLagWithMaxPollRecords")
val oldLag = consumer.metrics.get(new MetricName(tp + ".records-lag", "consumer-fetch-manager-metrics", "", oldTags))

val tags = new util.HashMap[String, String]()
tags.put("client-id", "testPerPartitionLagWithMaxPollRecords")
tags.put("topic", tp.topic())
tags.put("partition", String.valueOf(tp.partition()))
val lag = consumer.metrics.get(new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags))
assertEquals(oldLag.metricValue(), lag.metricValue())

assertEquals(s"The lag should be ${numMessages - records.count}", numMessages - records.count, lag.value, epsilon)
} finally {
consumer.close()
Expand Down
1 change: 1 addition & 0 deletions docs/upgrade.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ <h5><a id="upgrade_110_notable" href="#upgrade_110_notable">Notable changes in 1
<li>The kafka artifact in Maven no longer depends on log4j or slf4j-log4j12. Similarly to the kafka-clients artifact, users
can now choose the logging back-end by including the appropriate slf4j module (slf4j-log4j12, logback, etc.). The release
tarball still includes log4j and slf4j-log4j12.</li>
<li><a href="https://cwiki.apache.org/confluence/x/uaBzB">KIP-225</a> changed the metric "records.lag" to use tags for topic and partition. The original version with the name format "{topic}-{partition}.records-lag" is deprecated and will be removed in 2.0.0.</li>
</ul>

<h5><a id="upgrade_110_new_protocols" href="#upgrade_110_new_protocols">New Protocol Versions</a></h5>
Expand Down