You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "danoSF (via GitHub)" <gi...@apache.org> on 2025/08/22 18:41:28 UTC
[PR] @W-19339784 - Implementing KIP-977 [kafka]
danoSF opened a new pull request, #20397:
URL: https://github.com/apache/kafka/pull/20397
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.
--
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] Implementing KIP-977 [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20397:
URL: https://github.com/apache/kafka/pull/20397#issuecomment-4015485689
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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] Implementing KIP-977 [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20397:
URL: https://github.com/apache/kafka/pull/20397#issuecomment-3845149238
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496473371
##########
server/src/test/java/org/apache/kafka/server/metrics/MetricsVerbosityControllerTest.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.metrics;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class MetricsVerbosityControllerTest {
+
+ private static AbstractConfig configWith(String raw) {
+ ConfigDef def = new ConfigDef()
+ .define(MetricConfigs.METRICS_VERBOSITY_CONFIG,
+ ConfigDef.Type.STRING,
+ MetricConfigs.METRICS_VERBOSITY_DEFAULT,
+ ConfigDef.Importance.LOW,
+ MetricConfigs.METRICS_VERBOSITY_DOC);
+ return new AbstractConfig(def, Map.of(MetricConfigs.METRICS_VERBOSITY_CONFIG, raw));
+ }
+
+ @BeforeEach
+ public void resetCache() {
+ // Ensure a known baseline for the static cache between tests
+ AbstractConfig reset = configWith("[]");
+ MetricsVerbosityController.shouldEmitPartitionMetric(reset, "AnyMetric", "any-topic");
+ }
+
+ @Test
+ public void testEmptyConfigEmitsFalse() {
+ AbstractConfig conf = configWith("[]");
+ assertFalse(MetricsVerbosityController.shouldEmitPartitionMetric(conf, "BytesInPerSec", "car"));
+ }
+
+ @Test
+ public void testInvalidJsonEmitsFalse() {
+ AbstractConfig conf = configWith("not json");
+ assertFalse(MetricsVerbosityController.shouldEmitPartitionMetric(conf, "BytesInPerSec", "car"));
+ }
+
+ @Test
+ public void testLowLevelIgnored() {
+ String raw = "[ { \"level\": \"low\", \"names\": \"Bytes.*\", \"filters\": [{\"topics\":[\"car\"]}] } ]";
+ AbstractConfig conf = configWith(raw);
+ assertFalse(MetricsVerbosityController.shouldEmitPartitionMetric(conf, "BytesInPerSec", "car"));
+ }
+
+ @Test
+ public void testHighLevelNoTopicsDoesNotMatch() {
+ String raw = "[ { \"level\": \"high\", \"names\": \"Bytes.*\" } ]";
+ AbstractConfig conf = configWith(raw);
+ assertFalse(MetricsVerbosityController.shouldEmitPartitionMetric(conf, "BytesInPerSec", "car"));
+ }
+
+ @Test
+ public void testHighLevelWithTopicsAndNamePatternMatches() {
Review Comment:
Have you considered using JUnit 5 Parameterized Tests?
Example:
```java
@ParameterizedTest
@CsvSource({
"BytesInPerSec, car, true",
"BytesOutPerSec, bus, true",
"BytesInPerSec, plane, false",
"MessagesInPerSec, car, false"
})
public void testHighLevelWithTopicsAndNamePatternMatches(String metricName, String topic, boolean expected) {
String raw = "[ { \"level\": \"high\", \"names\": \"Bytes.*\", \"filters\": [{\"topics\":[\"car\",\"bus\"]}] } ]";
AbstractConfig conf = configWith(raw);
boolean result = MetricsVerbosityController.shouldEmitPartitionMetric(conf, metricName, topic);
assertEquals(expected, result,
String.format("Expected %s for metric '%s' on topic '%s'", expected, metricName, topic));
}
```
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496489782
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -4121,9 +4137,19 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.PRODUCE =>
brokerTopicStats.topicStats(tp.topic).produceMessageConversionsRate.mark(conversionCount)
brokerTopicStats.allTopicsStats.produceMessageConversionsRate.mark(conversionCount)
+ if (MetricsVerbosityController.shouldEmitPartitionMetric(
+ config, BrokerTopicMetrics.PRODUCE_MESSAGE_CONVERSIONS_PER_SEC, tp.topic)) {
+ val m = brokerTopicStats.partitionStats(tp.topic, tp.partition)
+ m.produceMessageConversionsRate().mark(conversionCount)
Review Comment:
What do you think of moving the `shouldEmitPartitionMetric` invokation to `MeterWrapper`?
The code here could simply invoke `m.produceMessageConversionsRate().mark(conversionCount)`.
`produceMessageConversionsRate()` could either return an empty implementor with a no-op `mark()` method, or the actual metric.
--
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] Implementing KIP-977 [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #20397: Implementing KIP-977
URL: https://github.com/apache/kafka/pull/20397
--
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] Implementing KIP-977 [kafka]
Posted by "afshing (via GitHub)" <gi...@apache.org>.
afshing commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2294691708
##########
clients/src/test/java/org/apache/kafka/clients/CommonClientConfigsTest.java:
##########
@@ -150,6 +150,8 @@ public void testMetricsReporters() {
assertEquals(2, reporters.size());
}
+
Review Comment:
Remove these extra lines
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -94,7 +94,7 @@ object DynamicBrokerConfig {
LogCleaner.RECONFIGURABLE_CONFIGS.asScala ++
DynamicLogConfig.ReconfigurableConfigs ++
DynamicThreadPool.RECONFIGURABLE_CONFIGS.asScala ++
- Set(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG) ++
+ Set(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG, MetricConfigs.METRICS_VERBOSITY_CONFIG) ++
Review Comment:
It seems like the convention is to add each dynamic config in a separate line. Maybe we should follow that
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4488,6 +4488,192 @@ class KafkaApisTest extends Logging {
assertEquals(Seq("broker2"), node.map(_.host))
}
+ @Test
+ def testPartitionLevelMetricsGatedByMetricsVerbosity_Fetch_KafkaApis(): Unit = {
+ val t = "t"
+ val u = "u"
+ val tId = Uuid.randomUuid()
+ val uId = Uuid.randomUuid()
+ val tpT = new TopicPartition(t, 0)
+ val tpU = new TopicPartition(u, 0)
+ val tidpT = new TopicIdPartition(tId, tpT)
+ val tidpU = new TopicIdPartition(uId, tpU)
+
+ addTopicToMetadataCache(t, 1, topicId = tId)
+ addTopicToMetadataCache(u, 1, topicId = uId)
+
+ // Ensure log configs exist (not strictly required for gating, but used by code paths)
+ when(replicaManager.getLogConfig(ArgumentMatchers.eq(tpT))).thenReturn(None)
+ when(replicaManager.getLogConfig(ArgumentMatchers.eq(tpU))).thenReturn(None)
+
+ // Mock replicaManager.fetchMessages to return bytes for both topics
+ when(replicaManager.fetchMessages(
+ any[FetchParams],
+ any[Seq[(TopicIdPartition, FetchRequest.PartitionData)]],
+ any[ReplicaQuota],
+ any[Seq[(TopicIdPartition, FetchPartitionData)] => Unit]()
+ )).thenAnswer(invocation => {
+ val callback = invocation.getArgument(3).asInstanceOf[Seq[(TopicIdPartition, FetchPartitionData)] => Unit]
+ val recT = MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("a".getBytes(StandardCharsets.UTF_8)))
+ val recU = MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("b".getBytes(StandardCharsets.UTF_8)))
+ callback(Seq(
+ tidpT -> new FetchPartitionData(Errors.NONE, 0, 0, recT,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false),
+ tidpU -> new FetchPartitionData(Errors.NONE, 0, 0, recU,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)
+ ))
+ })
+
+ // Build fetch contexts and request for both topics
+ val fetchData = util.Map.of(
+ tidpT, new FetchRequest.PartitionData(Uuid.ZERO_UUID, 0, 0, 1000, Optional.empty()),
+ tidpU, new FetchRequest.PartitionData(Uuid.ZERO_UUID, 0, 0, 1000, Optional.empty())
+ )
+ val fetchDataBuilder = util.Map.of(
+ tpT, new FetchRequest.PartitionData(Uuid.ZERO_UUID, 0, 0, 1000, Optional.empty()),
+ tpU, new FetchRequest.PartitionData(Uuid.ZERO_UUID, 0, 0, 1000, Optional.empty())
+ )
+ val fetchMetadata = new JFetchMetadata(0, 0)
+ val fetchContext = new FullFetchContext(time, new FetchSessionCacheShard(1000, 100),
+ fetchMetadata, fetchData, false, false)
+ when(fetchManager.newContext(
+ any[Short],
+ any[JFetchMetadata],
+ any[Boolean],
+ any[util.Map[TopicIdPartition, FetchRequest.PartitionData]],
+ any[util.List[TopicIdPartition]],
+ any[util.Map[Uuid, String]])).thenReturn(fetchContext)
+
+ // Configure metrics.verbosity to allow only t for BytesOut/TotalFetch
+ val overrideProps = Map(
+ "metrics.verbosity" -> "[{\"level\":\"high\",\"names\":\"BytesOutPerSec|TotalFetchRequestsPerSec\",\"filters\":[{\"topics\":[\"t\"]}]}]"
+ )
+ kafkaApis = createKafkaApis(overrideProperties = overrideProps)
+
+ // Baseline counts
+ val tPartMetrics = brokerTopicStats.partitionStats(t, 0)
+ val uPartMetrics = brokerTopicStats.partitionStats(u, 0)
+ val tBytesOutBefore = tPartMetrics.bytesOutRate().count()
+ val tTotalFetchBefore = tPartMetrics.totalFetchRequestRate().count()
+ val uBytesOutBefore = uPartMetrics.bytesOutRate().count()
+ val uTotalFetchBefore = uPartMetrics.totalFetchRequestRate().count()
+
+ val fetchRequest = new FetchRequest.Builder(ApiKeys.FETCH.latestVersion, ApiKeys.FETCH.latestVersion,
+ -1, -1, 100, 0, fetchDataBuilder).metadata(fetchMetadata).build()
+ val request = buildRequest(fetchRequest)
+ kafkaApis.handleFetchRequest(request)
+
+ verifyNoThrottling[FetchResponse](request)
+
+ // After fetch: topic t should advance; topic u should remain unchanged
Review Comment:
Something is strange here. First we check if bytesOutRate > bytesOutBefore, and then we check they are equal? Am I reading correct?
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4488,6 +4488,192 @@ class KafkaApisTest extends Logging {
assertEquals(Seq("broker2"), node.map(_.host))
}
+ @Test
Review Comment:
We also need tests for
- Produce
- the request from follower
- the request with size of zero
In general all the conditional branches
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -745,6 +746,22 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topicResponse.topic != null) {
val tp = new TopicIdPartition(topicResponse.topicId, new TopicPartition(topicResponse.topic, data.partitionIndex))
brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), FetchResponse.recordsSize(data))
+ // Partition-level throughput (KIP-977): record per-partition fetch metrics for leader (client fetches), gated by metrics.verbosity
+ if (!fetchRequest.isFromFollower) {
+ val size = FetchResponse.recordsSize(data)
+ val serverConfig = config
+ if (size > 0 && org.apache.kafka.server.metrics.MetricsVerbosityController.shouldEmitPartitionMetric(
+ serverConfig, BrokerTopicMetrics.BYTES_OUT_PER_SEC, tp.topic)) {
+ val m = brokerTopicStats.partitionStats(tp.topic, tp.partition)
+ m.bytesOutRate().mark(size)
+ }
+ if (org.apache.kafka.server.metrics.MetricsVerbosityController.shouldEmitPartitionMetric(
Review Comment:
Let's import `org.apache.kafka.server.metrics.MetricsVerbosityController` in this file and avoid package name
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -745,6 +746,22 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topicResponse.topic != null) {
val tp = new TopicIdPartition(topicResponse.topicId, new TopicPartition(topicResponse.topic, data.partitionIndex))
brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), FetchResponse.recordsSize(data))
+ // Partition-level throughput (KIP-977): record per-partition fetch metrics for leader (client fetches), gated by metrics.verbosity
+ if (!fetchRequest.isFromFollower) {
+ val size = FetchResponse.recordsSize(data)
+ val serverConfig = config
Review Comment:
It seems like we can pass the `config` as is, and there is no value in assigning it to a new variable
##########
server/src/main/java/org/apache/kafka/server/metrics/MetricConfigs.java:
##########
@@ -71,11 +78,15 @@ public class MetricConfigs {
.define(METRIC_SAMPLE_WINDOW_MS_CONFIG, LONG, METRIC_SAMPLE_WINDOW_MS_DEFAULT, atLeast(1), LOW, METRIC_SAMPLE_WINDOW_MS_DOC)
.define(METRIC_REPORTER_CLASSES_CONFIG, LIST, METRIC_REPORTER_CLASSES_DEFAULT, LOW, METRIC_REPORTER_CLASSES_DOC)
.define(METRIC_RECORDING_LEVEL_CONFIG, STRING, METRIC_RECORDING_LEVEL_DEFAULT, LOW, METRIC_RECORDING_LEVEL_DOC)
+ .define(METRICS_VERBOSITY_CONFIG, STRING, METRICS_VERBOSITY_DEFAULT, LOW, METRICS_VERBOSITY_DOC)
// Kafka Yammer Metrics Reporter Configuration
.define(KAFKA_METRICS_REPORTER_CLASSES_CONFIG, LIST, KAFKA_METRIC_REPORTER_CLASSES_DEFAULT, LOW, KAFKA_METRICS_REPORTER_CLASSES_DOC)
.define(KAFKA_METRICS_POLLING_INTERVAL_SECONDS_CONFIG, INT, KAFKA_METRICS_POLLING_INTERVAL_SECONDS_DEFAULT, atLeast(1), LOW, KAFKA_METRICS_POLLING_INTERVAL_SECONDS_DOC)
+ // Kafka Metrics Verbosity Configuration (KIP-977)
Review Comment:
No need for this comment either
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -245,6 +245,244 @@ class ReplicaManagerTest {
}
}
+ @Test
+ def testPartitionLevelMetricsGatedByMetricsVerbosity_Produce(): Unit = {
+ val dir = TestUtils.tempDir()
+ val props = TestUtils.createBrokerConfig(0)
+ // Enable KIP-977 partition-level Bytes* metrics only for topic "t1"
+ props.put("metrics.verbosity", "[{\"level\":\"high\",\"names\":\"Bytes.*\",\"filters\":[{\"topics\":[\"t1\"]}]}]")
+ val config = KafkaConfig.fromProps(props)
+ val logManager = TestUtils.createLogManager(Seq(dir).asJava)
+ val rm = new ReplicaManager(
+ metrics = metrics,
+ config = config,
+ time = time,
+ scheduler = new MockScheduler(time),
+ logManager = logManager,
+ quotaManagers = quotaManager,
+ metadataCache = metadataCache,
+ logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
+ alterPartitionManager = alterPartitionManager)
+ try {
+ val t1 = "t1"
+ val t2 = "t2"
+ val t1Id = Uuid.randomUuid()
+ val t2Id = Uuid.randomUuid()
+ setupMetadataCacheWithTopicIds(Map(t1 -> t1Id, t2 -> t2Id), metadataCache)
+ val tp1 = new TopicPartition(t1, 0)
+ val tp2 = new TopicPartition(t2, 0)
+
+ val leaderDelta = topicsCreateDelta(0, isStartIdLeader = true, partitions = List(0), topicName = t1, topicId = t1Id)
+ rm.applyDelta(leaderDelta, imageFromTopics(leaderDelta.apply()))
+ val leaderDelta2 = topicsCreateDelta(0, isStartIdLeader = true, partitions = List(0), topicName = t2, topicId = t2Id)
+ rm.applyDelta(leaderDelta2, imageFromTopics(leaderDelta2.apply()))
+
+ val t1PartMetrics = rm.brokerTopicStats.partitionStats(t1, 0)
+ val t2PartMetrics = rm.brokerTopicStats.partitionStats(t2, 0)
+
+ val t1BytesBefore = t1PartMetrics.bytesInRate().count()
+ val t1MsgsBefore = t1PartMetrics.messagesInRate().count()
+ val t2BytesBefore = t2PartMetrics.bytesInRate().count()
+ val t2MsgsBefore = t2PartMetrics.messagesInRate().count()
+
+ // Append to t1 should mark partition-level Bytes* and Messages* meters
+ appendRecords(rm, tp1, TestUtils.singletonRecords("a".getBytes)).onFire { response =>
+ assertEquals(Errors.NONE, response.error)
+ }
+ assertTrue(t1PartMetrics.bytesInRate().count() > t1BytesBefore)
Review Comment:
Can we be more specific and check if the count is actually equal to bytesBefore + diff?
##########
server/src/main/java/org/apache/kafka/server/metrics/MetricConfigs.java:
##########
@@ -46,6 +46,8 @@ public class MetricConfigs {
public static final String METRIC_RECORDING_LEVEL_DEFAULT = Sensor.RecordingLevel.INFO.toString();
public static final String METRIC_RECORDING_LEVEL_DOC = CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC;
+ // KIP-977: metrics verbosity config is defined in CommonClientConfigs for consistency
Review Comment:
This comment seems to be a left over
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496436811
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -669,6 +671,20 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topicResponse.topic != null) {
val tp = new TopicIdPartition(topicResponse.topicId, new TopicPartition(topicResponse.topic, data.partitionIndex))
brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), FetchResponse.recordsSize(data))
+ // KIP-977: Partition metrics for client fetches
+ if (!fetchRequest.isFromFollower) {
Review Comment:
What do you think of moving this logic to a new method in `brokerTopicStats`?
(e.g. `updateBytesOutPerPartition`)
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496423789
##########
clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java:
##########
@@ -132,6 +132,24 @@ public class CommonClientConfigs {
public static final String METRICS_CONTEXT_PREFIX = "metrics.context.";
+ public static final String METRICS_VERBOSITY_CONFIG = "metrics.verbosity";
+ public static final String METRICS_VERBOSITY_DOC = "Controls the verbosity of metrics reporting. This configuration accepts a JSON array string that specifies " +
+ "verbosity levels for specific metrics and scopes. Each array element should contain:\n" +
+ "- <code>level</code>: Verbosity level (\"high\", \"medium\", or \"low\")\n" +
+ "- <code>name</code>: Metric name pattern (supports regex)\n" +
+ "- <code>filters</code>: Scope filters to apply the verbosity level\n\n" +
+ "Example:\n" +
+ "<pre>\n" +
+ "[\n" +
+ " {\n" +
+ " \"level\": \"high\",\n" +
+ " \"name\": \"Bytes*\",\n" +
+ " \"filters\": [{\"topics\": [\"car\", \"bus\", \"plane\"]}]\n" +
+ " }\n" +
+ "]\n" +
+ "</pre>\n" +
+ "This example applies 'high' verbosity to all metrics starting with 'Bytes' for the specified topics (car, bus, plane).";
Review Comment:
Have you considered adding more information about the filters format, as they are described [here](https://cwiki.apache.org/confluence/display/KAFKA/KIP-977%3A+Partition-Level+Throughput+Metrics#KIP977:PartitionLevelThroughputMetrics-JSONConfigFormat) in the KIP?
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496489782
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -4121,9 +4137,19 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.PRODUCE =>
brokerTopicStats.topicStats(tp.topic).produceMessageConversionsRate.mark(conversionCount)
brokerTopicStats.allTopicsStats.produceMessageConversionsRate.mark(conversionCount)
+ if (MetricsVerbosityController.shouldEmitPartitionMetric(
+ config, BrokerTopicMetrics.PRODUCE_MESSAGE_CONVERSIONS_PER_SEC, tp.topic)) {
+ val m = brokerTopicStats.partitionStats(tp.topic, tp.partition)
+ m.produceMessageConversionsRate().mark(conversionCount)
Review Comment:
What do you think of moving the `shouldEmitPartitionMetric` invokation to `MeterWrapper`?
The code here could simply invoke `m.produceMessageConversionsRate().mark(conversionCount)`.
`produceMessageConversionsRate()` could either return an empty implementor with a no-op `mark()` method, or the actual meter.
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496436811
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -669,6 +671,20 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topicResponse.topic != null) {
val tp = new TopicIdPartition(topicResponse.topicId, new TopicPartition(topicResponse.topic, data.partitionIndex))
brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), FetchResponse.recordsSize(data))
+ // KIP-977: Partition metrics for client fetches
+ if (!fetchRequest.isFromFollower) {
Review Comment:
What do you think of moving this logic to a new method in `brokerTopicStats`?
--
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] Implementing KIP-977 [kafka]
Posted by "d4v1de (via GitHub)" <gi...@apache.org>.
d4v1de commented on code in PR #20397:
URL: https://github.com/apache/kafka/pull/20397#discussion_r2496436811
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -669,6 +671,20 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topicResponse.topic != null) {
val tp = new TopicIdPartition(topicResponse.topicId, new TopicPartition(topicResponse.topic, data.partitionIndex))
brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), FetchResponse.recordsSize(data))
+ // KIP-977: Partition metrics for client fetches
+ if (!fetchRequest.isFromFollower) {
Review Comment:
What do you think of moving this logic to a new method in `brokerTopicStats`?
Example:
```scala
val size = FetchResponse.recordsSize(data)
brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), size)
brokerTopicStats.updateBytesOutPerPartition(tp.topic, tp.partition, fetchRequest.isFromFollower, size)
```
--
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