KAFKA-17894: Implemented broker topic metrics for Share Group 1/N (KIP-1103) (apache#18444)

The PR implements the BrokerTopicMetrics defined in KIP-1103.

The PR also corrected the share-acknowledgement-rate and share-acknowledgement-count metrics defined in KIP-932 as they are moved to BrokerTopicMetrics, necessary changes to KIP-932 broker metrics will be done once we complete KIP-1103.

Reviewers: Andrew Schofield <[email protected]>, Chia-Ping Tsai <[email protected]>, Jun Rao <[email protected]>

Author: apoorvmittal10

checkstyle/import-control-server.xml

@@ -92,6 +92,9 @@
       <allow pkg="org.apache.kafka.network" />
       <allow pkg="org.apache.kafka.storage.internals.log" />
     </subpackage>
+    <subpackage name="share">
+      <allow pkg="org.apache.kafka.storage.log.metrics" />
+    </subpackage>
   </subpackage>
 
   <subpackage name="security">

checkstyle/suppressions.xml

@@ -45,7 +45,7 @@
               files="(KafkaClusterTestKit).java"/>
     <suppress checks="NPathComplexity" files="TestKitNodes.java"/>
     <suppress checks="JavaNCSS"
-              files="(RemoteLogManagerTest|SharePartitionTest).java"/>
+              files="(RemoteLogManagerTest|SharePartitionManagerTest|SharePartitionTest).java"/>
     <suppress checks="ClassDataAbstractionCoupling|ClassFanOutComplexity" files="SharePartitionManagerTest"/>
     <suppress checks="CyclomaticComplexity" files="SharePartition.java"/>
 

core/src/main/java/kafka/server/share/SharePartitionManager.java

@@ -60,6 +60,7 @@
 import org.apache.kafka.server.util.timer.SystemTimer;
 import org.apache.kafka.server.util.timer.SystemTimerReaper;
 import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -68,13 +69,17 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 
 /**
  * The SharePartitionManager is responsible for managing the SharePartitions and ShareSessions.
@@ -140,6 +145,11 @@ public class SharePartitionManager implements AutoCloseable {
      */
     private final ShareGroupMetrics shareGroupMetrics;
 
+    /**
+     * The broker topic stats is used to record the broker topic metrics for share group.
+     */
+    private final BrokerTopicStats brokerTopicStats;
+
     /**
      * The max fetch records is the maximum number of records that can be fetched by a share fetch request.
      */
@@ -155,7 +165,8 @@ public SharePartitionManager(
         int maxFetchRecords,
         Persister persister,
         GroupConfigManager groupConfigManager,
-        Metrics metrics
+        Metrics metrics,
+        BrokerTopicStats brokerTopicStats
     ) {
         this(replicaManager,
             time,
@@ -167,7 +178,8 @@ public SharePartitionManager(
             maxFetchRecords,
             persister,
             groupConfigManager,
-            metrics
+            metrics,
+            brokerTopicStats
         );
     }
 
@@ -182,7 +194,8 @@ private SharePartitionManager(
         int maxFetchRecords,
         Persister persister,
         GroupConfigManager groupConfigManager,
-        Metrics metrics
+        Metrics metrics,
+        BrokerTopicStats brokerTopicStats
     ) {
         this(replicaManager,
             time,
@@ -196,7 +209,8 @@ private SharePartitionManager(
             maxFetchRecords,
             persister,
             groupConfigManager,
-            metrics
+            metrics,
+            brokerTopicStats
         );
     }
 
@@ -213,7 +227,8 @@ private SharePartitionManager(
             int maxFetchRecords,
             Persister persister,
             GroupConfigManager groupConfigManager,
-            Metrics metrics
+            Metrics metrics,
+            BrokerTopicStats brokerTopicStats
     ) {
         this.replicaManager = replicaManager;
         this.time = time;
@@ -227,6 +242,7 @@ private SharePartitionManager(
         this.groupConfigManager = groupConfigManager;
         this.shareGroupMetrics = new ShareGroupMetrics(Objects.requireNonNull(metrics), time);
         this.maxFetchRecords = maxFetchRecords;
+        this.brokerTopicStats = brokerTopicStats;
     }
 
     /**
@@ -252,7 +268,7 @@ public CompletableFuture<Map<TopicIdPartition, PartitionData>> fetchMessages(
                 partitionMaxBytes.keySet(), groupId, fetchParams);
 
         CompletableFuture<Map<TopicIdPartition, PartitionData>> future = new CompletableFuture<>();
-        processShareFetch(new ShareFetch(fetchParams, groupId, memberId, future, partitionMaxBytes, batchSize, maxFetchRecords));
+        processShareFetch(new ShareFetch(fetchParams, groupId, memberId, future, partitionMaxBytes, batchSize, maxFetchRecords, brokerTopicStats));
 
         return future;
     }
@@ -274,9 +290,11 @@ public CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.Part
     ) {
         log.trace("Acknowledge request for topicIdPartitions: {} with groupId: {}",
             acknowledgeTopics.keySet(), groupId);
-        this.shareGroupMetrics.shareAcknowledgement();
         Map<TopicIdPartition, CompletableFuture<Throwable>> futures = new HashMap<>();
+        // Track the topics for which we have received an acknowledgement for metrics.
+        Set<String> topics = new HashSet<>();
         acknowledgeTopics.forEach((topicIdPartition, acknowledgePartitionBatches) -> {
+            topics.add(topicIdPartition.topic());
             SharePartitionKey sharePartitionKey = sharePartitionKey(groupId, topicIdPartition);
             SharePartition sharePartition = partitionCacheMap.get(sharePartitionKey);
             if (sharePartition != null) {
@@ -302,7 +320,13 @@ public CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.Part
             }
         });
 
-        return mapAcknowledgementFutures(futures);
+        // Update the metrics for the topics for which we have received an acknowledgement.
+        topics.forEach(topic -> {
+            brokerTopicStats.allTopicsStats().totalShareAcknowledgementRequestRate().mark();
+            brokerTopicStats.topicStats(topic).totalShareAcknowledgementRequestRate().mark();
+        });
+
+        return mapAcknowledgementFutures(futures, Optional.of(failedShareAcknowledgeMetricsHandler()));
     }
 
     /**
@@ -363,24 +387,31 @@ public CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.Part
             }
         });
 
-        return mapAcknowledgementFutures(futuresMap);
+        return mapAcknowledgementFutures(futuresMap, Optional.empty());
     }
 
-    private CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData>> mapAcknowledgementFutures(Map<TopicIdPartition, CompletableFuture<Throwable>> futuresMap) {
+    private CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData>> mapAcknowledgementFutures(
+        Map<TopicIdPartition, CompletableFuture<Throwable>> futuresMap,
+        Optional<Consumer<Set<String>>> failedMetricsHandler
+    ) {
         CompletableFuture<Void> allFutures = CompletableFuture.allOf(
             futuresMap.values().toArray(new CompletableFuture[0]));
         return allFutures.thenApply(v -> {
             Map<TopicIdPartition, ShareAcknowledgeResponseData.PartitionData> result = new HashMap<>();
+            // Keep the set as same topic might appear multiple times. Multiple partitions can fail for same topic.
+            Set<String> failedTopics = new HashSet<>();
             futuresMap.forEach((topicIdPartition, future) -> {
                 ShareAcknowledgeResponseData.PartitionData partitionData = new ShareAcknowledgeResponseData.PartitionData()
                     .setPartitionIndex(topicIdPartition.partition());
                 Throwable t = future.join();
                 if (t != null) {
                     partitionData.setErrorCode(Errors.forException(t).code())
                         .setErrorMessage(t.getMessage());
+                    failedTopics.add(topicIdPartition.topic());
                 }
                 result.put(topicIdPartition, partitionData);
             });
+            failedMetricsHandler.ifPresent(handler -> handler.accept(failedTopics));
             return result;
         });
     }
@@ -554,7 +585,10 @@ void processShareFetch(ShareFetch shareFetch) {
 
         List<DelayedShareFetchKey> delayedShareFetchWatchKeys = new ArrayList<>();
         LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
+        // Track the topics for which we have received a share fetch request for metrics.
+        Set<String> topics = new HashSet<>();
         for (TopicIdPartition topicIdPartition : shareFetch.partitionMaxBytes().keySet()) {
+            topics.add(topicIdPartition.topic());
             SharePartitionKey sharePartitionKey = sharePartitionKey(
                 shareFetch.groupId(),
                 topicIdPartition
@@ -598,6 +632,12 @@ void processShareFetch(ShareFetch shareFetch) {
             sharePartitions.put(topicIdPartition, sharePartition);
         }
 
+        // Update the metrics for the topics for which we have received a share fetch request.
+        topics.forEach(topic -> {
+            brokerTopicStats.allTopicsStats().totalShareFetchRequestRate().mark();
+            brokerTopicStats.topicStats(topic).totalShareFetchRequestRate().mark();
+        });
+
         // If all the partitions in the request errored out, then complete the fetch request with an exception.
         if (shareFetch.errorInAllPartitions()) {
             shareFetch.maybeComplete(Collections.emptyMap());
@@ -695,6 +735,21 @@ private static void removeSharePartitionFromCache(
         }
     }
 
+    /**
+     * The handler to update the failed share acknowledge request metrics.
+     *
+     * @return A Consumer that updates the failed share acknowledge request metrics.
+     */
+    private Consumer<Set<String>> failedShareAcknowledgeMetricsHandler() {
+        return failedTopics -> {
+            // Update failed share acknowledge request metric.
+            failedTopics.forEach(topic -> {
+                brokerTopicStats.allTopicsStats().failedShareAcknowledgementRequestRate().mark();
+                brokerTopicStats.topicStats(topic).failedShareAcknowledgementRequestRate().mark();
+            });
+        };
+    }
+
     /**
      * The SharePartitionListener is used to listen for partition events. The share partition is associated with
      * the topic-partition, we need to handle the partition events for the share partition.
@@ -759,10 +814,6 @@ static class ShareGroupMetrics {
 
         public static final String METRICS_GROUP_NAME = "share-group-metrics";
 
-        public static final String SHARE_ACK_SENSOR = "share-acknowledgement-sensor";
-        public static final String SHARE_ACK_RATE = "share-acknowledgement-rate";
-        public static final String SHARE_ACK_COUNT = "share-acknowledgement-count";
-
         public static final String RECORD_ACK_SENSOR_PREFIX = "record-acknowledgement";
         public static final String RECORD_ACK_RATE = "record-acknowledgement-rate";
         public static final String RECORD_ACK_COUNT = "record-acknowledgement-count";
@@ -775,7 +826,6 @@ static class ShareGroupMetrics {
         public static final Map<Byte, String> RECORD_ACKS_MAP = new HashMap<>();
 
         private final Time time;
-        private final Sensor shareAcknowledgementSensor;
         private final Map<Byte, Sensor> recordAcksSensorMap = new HashMap<>();
         private final Sensor partitionLoadTimeSensor;
 
@@ -787,18 +837,6 @@ static class ShareGroupMetrics {
 
         public ShareGroupMetrics(Metrics metrics, Time time) {
             this.time = time;
-
-            shareAcknowledgementSensor = metrics.sensor(SHARE_ACK_SENSOR);
-            shareAcknowledgementSensor.add(new Meter(
-                metrics.metricName(
-                    SHARE_ACK_RATE,
-                    METRICS_GROUP_NAME,
-                    "Rate of acknowledge requests."),
-                metrics.metricName(
-                    SHARE_ACK_COUNT,
-                    METRICS_GROUP_NAME,
-                    "The number of acknowledge requests.")));
-
             for (Map.Entry<Byte, String> entry : RECORD_ACKS_MAP.entrySet()) {
                 recordAcksSensorMap.put(entry.getKey(), metrics.sensor(String.format("%s-%s-sensor", RECORD_ACK_SENSOR_PREFIX, entry.getValue())));
                 recordAcksSensorMap.get(entry.getKey())
@@ -828,10 +866,6 @@ public ShareGroupMetrics(Metrics metrics, Time time) {
                 new Max());
         }
 
-        void shareAcknowledgement() {
-            shareAcknowledgementSensor.record();
-        }
-
         void recordAcknowledgement(byte ackType) {
             // unknown ack types (such as gaps for control records) are intentionally ignored
             if (recordAcksSensorMap.containsKey(ackType)) {

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -440,7 +440,8 @@ class BrokerServer(
         config.shareGroupConfig.shareFetchMaxFetchRecords,
         persister,
         groupConfigManager,
-        metrics
+        metrics,
+        brokerTopicStats
       )
 
       dataPlaneRequestProcessor = new KafkaApis(