Backend group details (#19)

* temp result commit

* consumer group details api done

* fixes

* changed calls from loop into vars, changed foreach to map

* changes

* removed redundant import

* Refactoring, fixed contract

* Fixed useless group query

* fix-consumer-groups-retaining-cluster-on-cluster-switch

Co-authored-by: Roman Nedzvetskiy <roman@Romans-MacBook-Pro.local>
Co-authored-by: German Osin <german.osin@gmail.com>
Co-authored-by: Azat Gataullin <gataniel@gmail.com>

kafka-ui-api/pom.xml

@@ -49,6 +49,11 @@
             <artifactId>kafka-clients</artifactId>
             <version>${kafka-clients.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.13</artifactId>
+            <version>${kafka.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.101tec</groupId>
             <artifactId>zkclient</artifactId>

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/ClusterService.java

@@ -8,12 +8,17 @@ import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.util.List;
+import java.util.*;
 import java.util.stream.Collectors;
 
 @Service
@@ -62,12 +67,55 @@ public class ClusterService {
         return kafkaService.createTopic(cluster, topicFormData);
     }
 
+    public Mono<ResponseEntity<ConsumerGroupDetails>> getConsumerGroupDetail(String clusterName, String consumerGroupId) {
+        KafkaCluster cluster = clustersStorage.getClusterByName(clusterName);
+
+        return ClusterUtil.toMono(
+                        cluster.getAdminClient()
+                                .describeConsumerGroups(Collections.singletonList(consumerGroupId)).all()
+                ).flatMap(groups ->
+                        groupMetadata(cluster, consumerGroupId).map(
+                            offsets -> {
+                                Map<TopicPartition, Long> endOffsets = topicPartitionsEndOffsets(cluster, offsets.keySet());
+                                return groups.get(consumerGroupId).members().stream()
+                                        .flatMap(c -> ClusterUtil.convertToConsumerTopicPartitionDetails(c, offsets, endOffsets).stream())
+                                        .collect(Collectors.toList());
+                            }
+                        )
+                )
+                .map(c -> new ConsumerGroupDetails().consumers(c).consumerGroupId(consumerGroupId))
+                .map(ResponseEntity::ok);
+
+    }
+
+    public Mono<Map<TopicPartition, OffsetAndMetadata>> groupMetadata(KafkaCluster cluster, String consumerGroupId) {
+        return ClusterUtil.toMono(
+                cluster.getAdminClient().listConsumerGroupOffsets(consumerGroupId).partitionsToOffsetAndMetadata()
+        );
+    }
+
+    public Map<TopicPartition, Long> topicPartitionsEndOffsets(KafkaCluster cluster, Collection<TopicPartition> topicPartitions) {
+        Map<TopicPartition, Long> result;
+        Properties properties = new Properties();
+        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers());
+        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
+
+        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
+            return consumer.endOffsets(topicPartitions);
+        }
+    }
+
     @SneakyThrows
-    public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroup (String clusterName) {
+    public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroups (String clusterName) {
             var cluster = clustersStorage.getClusterByName(clusterName);
             return ClusterUtil.toMono(cluster.getAdminClient().listConsumerGroups().all())
-                    .flatMap(s -> ClusterUtil.toMono(cluster.getAdminClient()
-                            .describeConsumerGroups(s.stream().map(ConsumerGroupListing::groupId).collect(Collectors.toList())).all()))
+                    .flatMap(s ->
+                            ClusterUtil.toMono(cluster.getAdminClient()
+                                .describeConsumerGroups(s.stream().map(ConsumerGroupListing::groupId).collect(Collectors.toList()))
+                                .all())
+                    )
                     .map(s -> s.values().stream()
                             .map(c -> ClusterUtil.convertToConsumerGroup(c, cluster)).collect(Collectors.toList()))
                     .map(s -> ResponseEntity.ok(Flux.fromIterable(s)));

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/util/ClusterUtil.java

@@ -2,12 +2,20 @@ package com.provectus.kafka.ui.cluster.util;
 
 import com.provectus.kafka.ui.cluster.model.KafkaCluster;
 import com.provectus.kafka.ui.model.ConsumerGroup;
+import com.provectus.kafka.ui.model.ConsumerTopicPartitionDetail;
+import com.provectus.kafka.ui.model.TopicPartitionDto;
 import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.clients.admin.MemberDescription;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
 import reactor.core.publisher.Mono;
 
 import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 public class ClusterUtil {
 
@@ -31,4 +39,25 @@ public class ClusterUtil {
         consumerGroup.setNumTopics(topics.size());
         return consumerGroup;
     }
+
+    public static List<ConsumerTopicPartitionDetail> convertToConsumerTopicPartitionDetails(
+            MemberDescription consumer,
+            Map<TopicPartition, OffsetAndMetadata> groupOffsets,
+            Map<TopicPartition, Long> endOffsets
+    ) {
+        return consumer.assignment().topicPartitions().stream()
+                .map(tp -> {
+                    Long currentOffset = groupOffsets.get(tp).offset();
+                    Long endOffset = endOffsets.get(tp);
+                    ConsumerTopicPartitionDetail cd = new ConsumerTopicPartitionDetail();
+                    cd.setConsumerId(consumer.consumerId());
+                    cd.setTopic(tp.topic());
+                    cd.setPartition(tp.partition());
+                    cd.setCurrentOffset(currentOffset);
+                    cd.setEndOffset(endOffset);
+                    cd.setMessagesBehind(endOffset - currentOffset);
+                    return cd;
+                }).collect(Collectors.toList());
+    }
+
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/MetricsRestController.java

@@ -4,7 +4,6 @@ import com.provectus.kafka.ui.api.ApiClustersApi;
 import com.provectus.kafka.ui.cluster.service.ClusterService;
 import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
-import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.server.ServerWebExchange;
@@ -56,7 +55,12 @@ public class MetricsRestController implements ApiClustersApi {
     }
 
     @Override
-    public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroup(String clusterName, ServerWebExchange exchange) {
-        return clusterService.getConsumerGroup(clusterName);
+    public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroups(String clusterName, ServerWebExchange exchange) {
+        return clusterService.getConsumerGroups(clusterName);
+    }
+
+    @Override
+    public Mono<ResponseEntity<ConsumerGroupDetails>> getConsumerGroup(String clusterName, String consumerGroupId, ServerWebExchange exchange) {
+        return clusterService.getConsumerGroupDetail(clusterName, consumerGroupId);
     }
 }

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -169,12 +169,37 @@ paths:
                 items:
                   $ref: '#/components/schemas/TopicConfig'
 
-  /api/clusters/{clusterName}/consumerGroups:
+  /api/clusters/{clusterName}/consumer-groups/{id}:
     get:
       tags:
         - /api/clusters
-      summary: getConsumerGroup
+      summary: get Consumer Group By Id
       operationId: getConsumerGroup
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: id
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ConsumerGroupDetails'
+
+  /api/clusters/{clusterName}/consumerGroups:
+    get:
+      tags:
+        - /api/clusters
+      summary: get all ConsumerGroups
+      operationId: getConsumerGroups
       parameters:
         - name: clusterName
           in: path
@@ -337,8 +362,45 @@ components:
         clusterId:
           type: string
         consumerGroupId:
-            type: string
+          type: string
         numConsumers:
-            type: integer
+          type: integer
         numTopics:
-            type: integer
+          type: integer
+
+    TopicPartitionDto:
+      type: object
+      properties:
+        topic:
+          type: string
+        partition:
+          type: integer
+      required:
+        - topic
+        - partition
+
+    ConsumerTopicPartitionDetail:
+      type: object
+      properties:
+        consumerId:
+          type: string
+        topic:
+          type: string
+        partition:
+          type: integer
+        currentOffset:
+          type: long
+        endOffset:
+          type: long
+        messagesBehind:
+          type: long
+
+    ConsumerGroupDetails:
+      type: object
+      properties:
+        consumerGroupId:
+          type: string
+        consumers:
+          type: array
+          items:
+            $ref: '#/components/schemas/ConsumerTopicPartitionDetail'

kafka-ui-react-app/src/redux/reducers/consumerGroups/selectors.ts

@@ -7,6 +7,7 @@ import { ConsumerGroupID, ConsumerGroupsState } from '../../interfaces/consumerG
 const consumerGroupsState = ({ consumerGroups }: RootState): ConsumerGroupsState => consumerGroups;
 
 const getConsumerGroupsMap = (state: RootState) => consumerGroupsState(state).byID;
+const getConsumerGroupsIDsList = (state: RootState) => consumerGroupsState(state).allIDs;
 
 const getConsumerGroupsListFetchingStatus = createFetchingSelector('GET_CONSUMER_GROUPS');
 const getConsumerGroupDetailsFetchingStatus = createFetchingSelector('GET_CONSUMER_GROUP_DETAILS');
@@ -24,11 +25,13 @@ export const getIsConsumerGroupDetailsFetched = createSelector(
 export const getConsumerGroupsList = createSelector(
   getIsConsumerGroupsListFetched,
   getConsumerGroupsMap,
-  (isFetched, byID) => {
+  getConsumerGroupsIDsList,
+  (isFetched, byID, ids) => {
     if (!isFetched) {
       return [];
     }
-    return Object.keys(byID).map( (key) => byID[key]);
+
+    return ids.map(key => byID[key]);
   },
 );
 

pom.xml

@@ -27,6 +27,7 @@
 		<openapi-generator-maven-plugin.version>4.2.2</openapi-generator-maven-plugin.version>
 		<swagger-annotations.version>1.6.0</swagger-annotations.version>
 		<springdoc-openapi-webflux-ui.version>1.2.32</springdoc-openapi-webflux-ui.version>
+		<kafka.version>2.4.1</kafka.version>
 	</properties>
 
 	<groupId>com.provectus</groupId>