WIP: Consumer groups list with search. (#17)

* Added concumer groups list with search.

* added endpoint for group consumers

* removed redundand code and imports

* changed method to async mono

* method located better

* changes after review

* changed foreach to map

Co-authored-by: Sofia Shnaidman <sshnaidman@provectus.com>
Co-authored-by: Roman Nedzvetskiy <roman@Romans-MacBook-Pro.local>

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

@@ -2,9 +2,12 @@ package com.provectus.kafka.ui.cluster.service;
 
 import com.provectus.kafka.ui.cluster.model.ClustersStorage;
 import com.provectus.kafka.ui.cluster.model.KafkaCluster;
+import com.provectus.kafka.ui.cluster.util.ClusterUtil;
 import com.provectus.kafka.ui.kafka.KafkaService;
 import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
@@ -58,4 +61,15 @@ public class ClusterService {
         if (cluster == null) return null;
         return kafkaService.createTopic(cluster, topicFormData);
     }
+
+    @SneakyThrows
+    public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroup (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()))
+                    .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

@@ -0,0 +1,34 @@
+package com.provectus.kafka.ui.cluster.util;
+
+import com.provectus.kafka.ui.cluster.model.KafkaCluster;
+import com.provectus.kafka.ui.model.ConsumerGroup;
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.common.KafkaFuture;
+import reactor.core.publisher.Mono;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class ClusterUtil {
+
+    public static <T> Mono<T> toMono(KafkaFuture<T> future){
+        return Mono.create(sink -> future.whenComplete((res, ex)->{
+            if (ex!=null) {
+                sink.error(ex);
+            } else {
+                sink.success(res);
+            }
+        }));
+    }
+
+    public static ConsumerGroup convertToConsumerGroup(ConsumerGroupDescription c, KafkaCluster cluster) {
+        ConsumerGroup consumerGroup = new ConsumerGroup();
+        consumerGroup.setClusterId(cluster.getCluster().getId());
+        consumerGroup.setConsumerGroupId(c.groupId());
+        consumerGroup.setNumConsumers(c.members().size());
+        Set<String> topics = new HashSet<>();
+        c.members().forEach(s1 -> s1.assignment().topicPartitions().forEach(s2 -> topics.add(s2.topic())));
+        consumerGroup.setNumTopics(topics.size());
+        return consumerGroup;
+    }
+}

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

@@ -4,6 +4,7 @@ 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;
@@ -53,4 +54,9 @@ public class MetricsRestController implements ApiClustersApi {
     public Mono<ResponseEntity<Flux<Broker>>> getBrokers(String clusterId, ServerWebExchange exchange) {
         return Mono.just(ResponseEntity.ok(Flux.fromIterable(new ArrayList<>())));
     }
+
+    @Override
+    public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroup(String clusterName, ServerWebExchange exchange) {
+        return clusterService.getConsumerGroup(clusterName);
+    }
 }

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

@@ -169,6 +169,28 @@ paths:
                 items:
                   $ref: '#/components/schemas/TopicConfig'
 
+  /api/clusters/{clusterName}/consumerGroups:
+    get:
+      tags:
+        - /api/clusters
+      summary: getConsumerGroup
+      operationId: getConsumerGroup
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/ConsumerGroup'
+
 components:
   schemas:
     Cluster:
@@ -307,4 +329,16 @@ components:
       type: object
       properties:
         id:
-          type: string
+          type: string
+
+    ConsumerGroup:
+      type: object
+      properties:
+        clusterId:
+          type: string
+        consumerGroupId:
+            type: string
+        numConsumers:
+            type: integer
+        numTopics:
+            type: integer

kafka-ui-react-app/mock/index.js

@@ -5,6 +5,7 @@ const brokerMetrics = require('./payload/brokerMetrics.json');
 const topics = require('./payload/topics.json');
 const topicDetails = require('./payload/topicDetails.json');
 const topicConfigs = require('./payload/topicConfigs.json');
+const consumerGroups = require('./payload/consumerGroups.json');
 
 const db = {
     clusters,
@@ -13,6 +14,7 @@ const db = {
     topics: topics.map((topic) => ({...topic, id: topic.name})),
     topicDetails,
     topicConfigs,
+    consumerGroups: consumerGroups.map((group) => ({...group, id: group.consumerGroupId}))
 };
 const server = jsonServer.create();
 const router = jsonServer.router(db);

kafka-ui-react-app/mock/payload/consumerGroups.json

@@ -0,0 +1,39 @@
+[
+  {
+    "clusterId": "fake.cluster",
+    "consumerGroupId": "_fake.cluster.consumer_1",
+    "numConsumers": 1,
+    "numTopics": 11
+  },
+  {
+    "clusterId": "fake.cluster",
+    "consumerGroupId": "_fake.cluster.consumer_2",
+    "numConsumers": 2,
+    "numTopics": 22
+  },
+  {
+    "clusterId": "fake.cluster",
+    "consumerGroupId": "_fake.cluster.consumer_3",
+    "numConsumers": 3,
+    "numTopics": 33
+  },
+
+  {
+    "clusterId": "kafka-ui.cluster",
+    "consumerGroupId": "_kafka-ui.cluster.consumer_1",
+    "numConsumers": 4,
+    "numTopics": 44
+  },
+  {
+    "clusterId": "kafka-ui.cluster",
+    "consumerGroupId": "_kafka-ui.cluster.consumer_2",
+    "numConsumers": 5,
+    "numTopics": 55
+  },
+  {
+    "clusterId": "kafka-ui.cluster",
+    "consumerGroupId": "_kafka-ui.cluster.consumer_3",
+    "numConsumers": 6,
+    "numTopics": 66
+  }
+]

kafka-ui-react-app/src/components/App.tsx

@@ -10,6 +10,7 @@ import TopicsContainer from './Topics/TopicsContainer';
 import NavConatiner from './Nav/NavConatiner';
 import PageLoader from './common/PageLoader/PageLoader';
 import Dashboard from './Dashboard/Dashboard';
+import ConsumersGroupsContainer from './ConsumerGroups/ConsumersGroupsContainer';
 
 interface AppProps {
   isClusterListFetched: boolean;
@@ -39,6 +40,7 @@ const App: React.FC<AppProps> = ({
             <Route exact path="/clusters" component={Dashboard} />
             <Route path="/clusters/:clusterName/topics" component={TopicsContainer} />
             <Route path="/clusters/:clusterName/brokers" component={BrokersContainer} />
+            <Route path="/clusters/:clusterName/consumer-groups" component={ConsumersGroupsContainer} />
             <Redirect from="/clusters/:clusterName" to="/clusters/:clusterName/brokers" />
           </Switch>
         ) : (

kafka-ui-react-app/src/components/ConsumerGroups/ConsumerGroups.tsx

@@ -0,0 +1,34 @@
+import React from 'react';
+import { ClusterName } from 'redux/interfaces';
+import {
+  Switch,
+  Route,
+} from 'react-router-dom';
+import ListContainer from './List/ListContainer';
+import PageLoader from 'components/common/PageLoader/PageLoader';
+
+interface Props {
+  clusterName: ClusterName;
+  isFetched: boolean;
+  fetchConsumerGroupsList: (clusterName: ClusterName) => void;
+}
+
+const ConsumerGroups: React.FC<Props> = ({
+  clusterName,
+  isFetched,
+  fetchConsumerGroupsList,
+}) => {
+  React.useEffect(() => { fetchConsumerGroupsList(clusterName); }, [fetchConsumerGroupsList, clusterName]);
+
+  if (isFetched) {
+    return (
+      <Switch>
+        <Route exact path="/clusters/:clusterName/consumer-groups" component={ListContainer} />
+      </Switch>
+    );
+  }
+
+  return (<PageLoader />);
+};
+
+export default ConsumerGroups;

kafka-ui-react-app/src/components/ConsumerGroups/ConsumersGroupsContainer.ts

@@ -0,0 +1,24 @@
+import { connect } from 'react-redux';
+import { fetchConsumerGroupsList } from 'redux/actions';
+import { RootState, ClusterName } from 'redux/interfaces';
+import { RouteComponentProps } from 'react-router-dom';
+import ConsumerGroups from './ConsumerGroups';
+import { getIsConsumerGroupsListFetched } from '../../redux/reducers/consumerGroups/selectors';
+
+
+interface RouteProps {
+  clusterName: ClusterName;
+}
+
+interface OwnProps extends RouteComponentProps<RouteProps> { }
+
+const mapStateToProps = (state: RootState, { match: { params: { clusterName } }}: OwnProps) => ({
+  isFetched: getIsConsumerGroupsListFetched(state),
+  clusterName,
+});
+
+const mapDispatchToProps = {
+  fetchConsumerGroupsList: (clusterName: ClusterName) => fetchConsumerGroupsList(clusterName),
+};
+
+export default connect(mapStateToProps, mapDispatchToProps)(ConsumerGroups);

kafka-ui-react-app/src/components/ConsumerGroups/List/List.tsx

@@ -0,0 +1,64 @@
+import React, { ChangeEvent } from 'react';
+import { ConsumerGroup, ClusterName } from 'redux/interfaces';
+import ListItem from './ListItem';
+import Breadcrumb from 'components/common/Breadcrumb/Breadcrumb';
+
+interface Props {
+  clusterName: ClusterName;
+  consumerGroups: (ConsumerGroup)[];
+}
+
+const List: React.FC<Props> = ({
+  consumerGroups,
+}) => {
+
+  const [searchText, setSearchText] = React.useState<string>('');
+
+  const handleInputChange = (event: React.ChangeEvent<HTMLInputElement>) => {
+    setSearchText(event.target.value);
+  };
+
+  const items = consumerGroups;
+
+  return (
+    <div className="section">
+      <Breadcrumb>All Consumer Groups</Breadcrumb>
+
+      <div className="box">
+        <div className="columns">
+          <div className="column is-half is-offset-half">
+            <input  id="searchText"
+                  type="text"
+                  name="searchText"
+                  className="input"
+                  placeholder="Search"
+                  value={searchText}
+                  onChange={handleInputChange}
+                />
+          </div>
+        </div>
+        <table className="table is-striped is-fullwidth">
+          <thead>
+            <tr>
+              <th>Consumer group ID</th>
+              <th>Num of consumers</th>
+              <th>Num of topics</th>
+            </tr>
+          </thead>
+          <tbody>
+            {items
+              .filter( (consumerGroup) => !searchText || consumerGroup?.consumerGroupId?.indexOf(searchText) >= 0)
+              .map((consumerGroup, index) => (
+                <ListItem
+                  key={`consumer-group-list-item-key-${index}`}
+                  {...consumerGroup}
+                />
+            ))}
+          </tbody>
+        </table>
+      </div>
+    </div>
+  );
+};
+
+export default List;

kafka-ui-react-app/src/components/ConsumerGroups/List/ListContainer.ts

@@ -0,0 +1,20 @@
+import { connect } from 'react-redux';
+import {ClusterName, RootState} from 'redux/interfaces';
+import { getConsumerGroupsList } from 'redux/reducers/consumerGroups/selectors';
+import List from './List';
+import { withRouter, RouteComponentProps } from 'react-router-dom';
+
+interface RouteProps {
+  clusterName: ClusterName;
+}
+
+interface OwnProps extends RouteComponentProps<RouteProps> { }
+
+const mapStateToProps = (state: RootState, { match: { params: { clusterName } } }: OwnProps) => ({
+  clusterName,
+  consumerGroups: getConsumerGroupsList(state)
+});
+
+export default withRouter(
+  connect(mapStateToProps)(List)
+);

kafka-ui-react-app/src/components/ConsumerGroups/List/ListItem.tsx

@@ -0,0 +1,24 @@
+import React from 'react';
+import { NavLink } from 'react-router-dom';
+import { ConsumerGroup } from 'redux/interfaces';
+
+const ListItem: React.FC<ConsumerGroup> = ({
+  consumerGroupId,
+  numConsumers,
+  numTopics,
+}) => {
+  return (
+    <tr>
+      {/* <td>
+        <NavLink exact to={`consumer-groups/${consumerGroupId}`} activeClassName="is-active" className="title is-6">
+          {consumerGroupId}
+        </NavLink>
+      </td> */}
+      <td>{consumerGroupId}</td>
+      <td>{numConsumers}</td>
+      <td>{numTopics}</td>
+    </tr>
+  );
+}
+
+export default ListItem;

kafka-ui-react-app/src/components/Nav/ClusterMenu.tsx

@@ -1,7 +1,7 @@
 import React, { CSSProperties } from 'react';
 import { Cluster } from 'redux/interfaces';
 import { NavLink } from 'react-router-dom';
-import { clusterBrokersPath, clusterTopicsPath } from 'lib/paths';
+import { clusterBrokersPath, clusterTopicsPath, clusterConsumerGroupsPath } from 'lib/paths';
 
 interface Props extends Cluster {}
 
@@ -37,6 +37,9 @@ const ClusterMenu: React.FC<Props> = ({
         <NavLink to={clusterTopicsPath(name)} activeClassName="is-active" title="Topics">
           Topics
         </NavLink>
+        <NavLink to={clusterConsumerGroupsPath(name)} activeClassName="is-active" title="Consumers">
+          Consumers
+        </NavLink>
       </ul>
     </li>
   </ul>

kafka-ui-react-app/src/lib/paths.ts

@@ -13,3 +13,5 @@ export const clusterTopicNewPath = (clusterName: ClusterName) => `${clusterPath(
 export const clusterTopicPath = (clusterName: ClusterName, topicName: TopicName) => `${clusterTopicsPath(clusterName)}/${topicName}`;
 export const clusterTopicSettingsPath = (clusterName: ClusterName, topicName: TopicName) => `${clusterTopicsPath(clusterName)}/${topicName}/settings`;
 export const clusterTopicMessagesPath = (clusterName: ClusterName, topicName: TopicName) => `${clusterTopicsPath(clusterName)}/${topicName}/messages`;
+
+export const clusterConsumerGroupsPath = (clusterName: ClusterName) => `${clusterPath(clusterName)}/consumer-groups`;

kafka-ui-react-app/src/redux/actionType.ts

@@ -26,4 +26,8 @@ export enum ActionType {
   POST_TOPIC__REQUEST = 'POST_TOPIC__REQUEST',
   POST_TOPIC__SUCCESS = 'POST_TOPIC__SUCCESS',
   POST_TOPIC__FAILURE = 'POST_TOPIC__FAILURE',
+
+  GET_CONSUMER_GROUPS__REQUEST = 'GET_CONSUMER_GROUPS__REQUEST',
+  GET_CONSUMER_GROUPS__SUCCESS = 'GET_CONSUMER_GROUPS__SUCCESS',
+  GET_CONSUMER_GROUPS__FAILURE = 'GET_CONSUMER_GROUPS__FAILURE',
 };

kafka-ui-react-app/src/redux/actions/actions.ts

@@ -1,5 +1,6 @@
 import { createAsyncAction} from 'typesafe-actions';
 import { ActionType } from 'redux/actionType';
+import { ConsumerGroup } from '../interfaces/consumerGroup';
 import {
   Broker,
   BrokerMetrics,
@@ -51,3 +52,9 @@ export const createTopicAction = createAsyncAction(
   ActionType.POST_TOPIC__SUCCESS,
   ActionType.POST_TOPIC__FAILURE,
 )<undefined, Topic, undefined>();
+
+export const fetchConsumerGroupsAction = createAsyncAction(
+  ActionType.GET_CONSUMER_GROUPS__REQUEST,
+  ActionType.GET_CONSUMER_GROUPS__SUCCESS,
+  ActionType.GET_CONSUMER_GROUPS__FAILURE,
+)<undefined, ConsumerGroup[], undefined>();

kafka-ui-react-app/src/redux/actions/thunks.ts

@@ -77,3 +77,13 @@ export const createTopic = (clusterName: ClusterName, form: TopicFormData): Prom
     dispatch(actions.createTopicAction.failure());
   }
 };
+
+export const fetchConsumerGroupsList = (clusterName: ClusterName): PromiseThunk<void> => async (dispatch) => {
+  dispatch(actions.fetchConsumerGroupsAction.request());
+  try {
+    const consumerGroups = await api.getConsumerGroups(clusterName);
+    dispatch(actions.fetchConsumerGroupsAction.success(consumerGroups));
+  } catch (e) {
+    dispatch(actions.fetchConsumerGroupsAction.failure());
+  }
+};

kafka-ui-react-app/src/redux/api/consumerGroups.ts

@@ -0,0 +1,8 @@
+import { ClusterName } from '../interfaces/cluster';
+import { ConsumerGroup } from '../interfaces/consumerGroup';
+import { BASE_PARAMS, BASE_URL } from '../../lib/constants';
+
+
+export const getConsumerGroups = (clusterName: ClusterName): Promise<ConsumerGroup[]> =>
+  fetch(`${BASE_URL}/clusters/${clusterName}/consumerGroups`, { ...BASE_PARAMS })
+    .then(res => res.json());

kafka-ui-react-app/src/redux/api/index.ts

@@ -1,3 +1,4 @@
 export * from './topics';
 export * from './clusters';
 export * from './brokers';
+export * from './consumerGroups';

kafka-ui-react-app/src/redux/interfaces/consumerGroup.ts

@@ -0,0 +1,5 @@
+export interface ConsumerGroup {
+  consumerGroupId: string;
+  numConsumers: number;
+  numTopics: number;
+}

kafka-ui-react-app/src/redux/interfaces/index.ts

@@ -8,10 +8,12 @@ import { TopicsState } from './topic';
 import { Cluster } from './cluster';
 import { BrokersState } from './broker';
 import { LoaderState } from './loader';
+import { ConsumerGroup } from './consumerGroup';
 
 export * from './topic';
 export * from './cluster';
 export * from './broker';
+export * from './consumerGroup';
 export * from './loader';
 
 export enum FetchStatus {
@@ -25,6 +27,7 @@ export interface RootState {
   topics: TopicsState;
   clusters: Cluster[];
   brokers: BrokersState;
+  consumerGroups: ConsumerGroup[];
   loader: LoaderState;
 }
 

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

@@ -0,0 +1,15 @@
+import { Action, ConsumerGroup } from 'redux/interfaces';
+import { ActionType } from 'redux/actionType';
+
+export const initialState: ConsumerGroup[] = [];
+
+const reducer = (state = initialState, action: Action): ConsumerGroup[] => {
+  switch (action.type) {
+    case ActionType.GET_CONSUMER_GROUPS__SUCCESS:
+      return action.payload;
+    default:
+      return state;
+  }
+};
+
+export default reducer;

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

@@ -0,0 +1,15 @@
+import { createSelector } from 'reselect';
+import { ConsumerGroup, RootState, FetchStatus } from 'redux/interfaces';
+import { createFetchingSelector } from 'redux/reducers/loader/selectors';
+
+
+const consumerGroupsState = ({ consumerGroups }: RootState): ConsumerGroup[] => consumerGroups;
+
+const getConsumerGroupsListFetchingStatus = createFetchingSelector('GET_CONSUMER_GROUPS');
+
+export const getIsConsumerGroupsListFetched = createSelector(
+  getConsumerGroupsListFetchingStatus,
+  (status) => status === FetchStatus.fetched,
+);
+
+export const getConsumerGroupsList = createSelector(consumerGroupsState, (consumerGroups) => consumerGroups);

kafka-ui-react-app/src/redux/reducers/index.ts

@@ -2,6 +2,7 @@ import { combineReducers } from 'redux';
 import topics from './topics/reducer';
 import clusters from './clusters/reducer';
 import brokers from './brokers/reducer';
+import consumerGroups from './consumerGroups/reducer';
 import loader from './loader/reducer';
 import { RootState } from 'redux/interfaces';
 
@@ -9,5 +10,6 @@ export default combineReducers<RootState>({
   topics,
   clusters,
   brokers,
+  consumerGroups,
   loader,
 });