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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
