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 2025/02/06 22:50:38 UTC

[PR] KAFKA-18733: Updating share group metrics (1/N) [kafka]

apoorvmittal10 opened a new pull request, #18826:
URL: https://github.com/apache/kafka/pull/18826

The PR does the following:
- Refactors ShareGroupMetrics out of SharePartitionManager as there are additional metrics need to be defined, as per KIP-1103, where ShareGroupMetrics is required to be pass to DelayedShareFetch, etc.
- As the KIP-1103 defines metrics which are Histogram in ShareGroupMetrics, also the BrokerTopicMetrics are Yammer metrics hence to maintian consistency, I moved existing ShareGroupMetrics (defined in KIP-932) also to Yammer metrics following Yammer metrics group naming convention i.e. from share-group-metrics to ShareGroupMetrics (this will require KIP-932 upadate where `share-acknowledgement` metric is already moved to BrokerTopicMetrics and other 2 in this PR to Yammer metrics).
- Moved `record-acknowledgement-rate|count` to Yammer meter metric with name `RecordAcknowledgementPerSec`.
- Moved `partition-load-time-avg|max` to Yammer histogram metric with name `PartitionLoadTimeMs`.
- Corrects the way partition load time metrics was calculated. The load time of share partition should be considered when the state of share partition moves from `empty` which happens once mayBeInitialize finishes. The existing implementation is not much helpful now since share partition initialization is now async, at the time of this metric implementation the initialization of share partition was sync hence the metric needed correction.

This PR is first step to start implementing other ShareGroupMetrics as defined in KIP-1103.

--
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-18733: Updating share group metrics (1/N) [kafka]

Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.

AndrewJSchofield commented on code in PR #18826:
URL: https://github.com/apache/kafka/pull/18826#discussion_r1950997858


##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -2172,14 +2124,88 @@ public void testPendingInitializationShouldCompleteFetchRequest() throws Excepti
         Mockito.verify(mockReplicaManager, times(0)).readFromLog(
             any(), any(), any(ReplicaQuota.class), anyBoolean());
         assertFalse(pendingInitializationFuture.isDone());
+        assertEquals(0, shareGroupMetrics.partitionLoadTimeMs().count());
         // Complete the pending initialization future.
         pendingInitializationFuture.complete(null);
+        // Verify the partition load time metrics.
+        assertEquals(1, shareGroupMetrics.partitionLoadTimeMs().count());
+        assertEquals(90.0, shareGroupMetrics.partitionLoadTimeMs().min());
+        assertEquals(90.0, shareGroupMetrics.partitionLoadTimeMs().max());
+        assertEquals(90.0, shareGroupMetrics.partitionLoadTimeMs().sum());
         // Should have 1 fetch recorded.
         validateBrokerTopicStatsMetrics(
             brokerTopicStats,
             new TopicMetrics(1, 0, 0, 0),
             Map.of(tp0.topic(), new TopicMetrics(1, 0, 0, 0))
         );
+        shareGroupMetrics.close();
+    }
+
+    @Test
+    public void testPartitionLoadTimeMetricWithMultiplePartitions() throws Exception {
+        String groupId = "grp";
+        TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+        TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
+        LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes = orderedMap(PARTITION_MAX_BYTES, tp0, tp1);
+
+        SharePartition sp0 = mock(SharePartition.class);
+        SharePartition sp1 = mock(SharePartition.class);
+        Map<SharePartitionKey, SharePartition> partitionCacheMap = new HashMap<>();
+        partitionCacheMap.put(new SharePartitionKey(groupId, tp0), sp0);
+        partitionCacheMap.put(new SharePartitionKey(groupId, tp1), sp1);
+
+        // Keep the initialization future pending, so fetch request is stuck.
+        CompletableFuture<Void> pendingInitializationFuture1 = new CompletableFuture<>();
+        when(sp0.maybeInitialize()).thenReturn(pendingInitializationFuture1);
+        when(sp0.loadStartTimeMs()).thenReturn(10L);
+
+        CompletableFuture<Void> pendingInitializationFuture2 = new CompletableFuture<>();
+        when(sp1.maybeInitialize()).thenReturn(pendingInitializationFuture2);
+        when(sp1.loadStartTimeMs()).thenReturn(40L);
+
+        DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(

Review Comment:
   I would expect a call to `delayedShareFetchPurgatory.shutdown()`, ideally whether the test fails or not. Because the reaper is not enabled, I suppose the risk of leaking a thread is not present, but it would still be preferable to leave things tidy.



##########
server/src/main/java/org/apache/kafka/server/share/metrics/ShareGroupMetrics.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.share.metrics;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * ShareGroupMetrics is used to track the broker-side metrics for the ShareGroup.
+ */
+public class ShareGroupMetrics implements AutoCloseable {
+    // Rate of records acknowledged per acknowledgement type.
+    private static final String RECORD_ACKNOWLEDGEMENTS_PER_SEC = "RecordAcknowledgementPerSec";
+    // The time in milliseconds to load the share partitions.
+    private static final String PARTITION_LOAD_TIME_MS = "PartitionLoadTimeMs";
+    private static final String ACK_TYPE = "ackType";

Review Comment:
   I suggest renaming this because it's specifically a tag. Maybe "ACK_TYPE_TAG".



##########
server/src/main/java/org/apache/kafka/server/share/metrics/ShareGroupMetrics.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.share.metrics;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * ShareGroupMetrics is used to track the broker-side metrics for the ShareGroup.
+ */
+public class ShareGroupMetrics implements AutoCloseable {
+    // Rate of records acknowledged per acknowledgement type.
+    private static final String RECORD_ACKNOWLEDGEMENTS_PER_SEC = "RecordAcknowledgementPerSec";

Review Comment:
   The string should be `"RecordAcknowledgementsPerSec"` (plural not singular). I think it's just a typo in the string name.



-- 
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-18733: Updating share group metrics (1/N) [kafka]

Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.

AndrewJSchofield merged PR #18826:
URL: https://github.com/apache/kafka/pull/18826


-- 
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