Merge pull request #4182 from nirga/multi-cluster

Kafka Plugin: Multi-cluster and multi-consumer per component

.changeset/fast-flowers-tickle.md

@@ -0,0 +1,31 @@
+---
+'@backstage/plugin-kafka': minor
+'@backstage/plugin-kafka-backend': minor
+---
+
+Added support for multiple Kafka clusters and multiple consumers per component.
+Note that this introduces several breaking changes.
+
+1. Configuration in `app-config.yaml` has changed to support the ability to configure multiple clusters. This means you are required to update the configs in the following way:
+
+```diff
+kafka:
+   clientId: backstage
+-  brokers:
+-    - localhost:9092
++  clusters:
++    - name: prod
++      brokers:
++        - localhost:9092
+```
+
+2. Configuration of services has changed as well to support multiple clusters:
+
+```diff
+  annotations:
+-    kafka.apache.org/consumer-groups: consumer
++    kafka.apache.org/consumer-groups: prod/consumer
+```
+
+3. Kafka Backend API has changed, so querying offsets of a consumer group is now done with the following query path:
+   `/consumers/${clusterId}/${consumerGroup}/offsets`

app-config.yaml

@@ -99,6 +99,13 @@ kubernetes:
     - 'config'
   clusters: []
 
+kafka:
+  clientId: backstage
+  clusters:
+    - name: cluster
+      brokers:
+        - localhost:9092
+
 integrations:
   github:
     - host: github.com
@@ -372,8 +379,3 @@ homepage:
       timezone: 'Asia/Tokyo'
 pagerduty:
   eventsBaseUrl: 'https://events.pagerduty.com/v2'
-
-kafka:
-  clientId: backstage
-  brokers:
-    - localhost:9092

plugins/kafka-backend/README.md

@@ -23,6 +23,8 @@ Example:
 ```yaml
 kafka:
   clientId: backstage
-  brokers:
-    - localhost:9092
+  clusters:
+    name: prod
+    brokers:
+      - localhost:9092
 ```

plugins/kafka-backend/config.d.ts

@@ -19,20 +19,22 @@ export interface Config {
      * Client ID used to Backstage uses to identify when connecting to the Kafka cluster.
      */
     clientId: string;
-    /**
-     * List of brokers in the Kafka cluster to connect to.
-     */
-    brokers: string[];
-
-    /**
-     * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect.
-     * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options
-     */
-    ssl?: {
-      ca: string[];
-      /** @visibility secret */
-      key: string;
-      cert: string;
-    };
+    clusters: {
+      name: string;
+      /**
+       * List of brokers in the Kafka cluster to connect to.
+       */
+      brokers: string[];
+      /**
+       * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect.
+       * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options
+       */
+      ssl?: {
+        ca: string[];
+        /** @visibility secret */
+        key: string;
+        cert: string;
+      };
+    }[];
   };
 }

plugins/kafka-backend/src/config/ClusterReader.test.ts

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2020 Spotify AB
+ *
+ * Licensed 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.
+ */
+
+import '@backstage/backend-common';
+import { ConfigReader } from '@backstage/config';
+import { getClusterDetails } from './ClusterReader';
+
+describe('getClusterDetails', () => {
+  it('empty clusters return empty cluster details', async () => {
+    const config = new ConfigReader({ clusters: [] });
+
+    const result = getClusterDetails(config.getConfigArray('clusters'));
+
+    expect(result).toStrictEqual([]);
+  });
+
+  it('two clusters return two cluster details', async () => {
+    const config = new ConfigReader({
+      clusters: [
+        { name: 'cluster1', brokers: ['a', 'b'] },
+        { name: 'cluster2', brokers: ['d'] },
+      ],
+    });
+
+    const result = getClusterDetails(config.getConfigArray('clusters'));
+
+    expect(result).toStrictEqual([
+      { name: 'cluster1', brokers: ['a', 'b'] },
+      { name: 'cluster2', brokers: ['d'] },
+    ]);
+  });
+});

plugins/kafka-backend/src/config/ClusterReader.ts

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020 Spotify AB
+ *
+ * Licensed 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.
+ */
+
+import { Config } from '@backstage/config';
+import { ConnectionOptions } from 'tls';
+import { ClusterDetails } from '../types/types';
+
+export function getClusterDetails(config: Config[]): ClusterDetails[] {
+  return config.map(clusterConfig => {
+    const clusterDetails = {
+      name: clusterConfig.getString('name'),
+      brokers: clusterConfig.getStringArray('brokers'),
+    };
+    const sslConfig = clusterConfig.getOptional('kafka.ssl');
+    if (sslConfig) {
+      return {
+        ...clusterDetails,
+        ssl: sslConfig as ConnectionOptions,
+      };
+    }
+
+    return clusterDetails;
+  });
+}

plugins/kafka-backend/src/service/router.test.ts

@@ -16,22 +16,34 @@
 
 import request from 'supertest';
 import express from 'express';
-import { makeRouter } from './router';
+import { makeRouter, ClusterApi } from './router';
 import { getVoidLogger } from '@backstage/backend-common';
 import { KafkaApi } from './KafkaApi';
 import { when } from 'jest-when';
 
 describe('router', () => {
   let app: express.Express;
-  let kafkaApi: jest.Mocked<KafkaApi>;
+  let apis: ClusterApi[];
+  let devKafkaApi: jest.Mocked<KafkaApi>;
+  let prodKafkaApi: jest.Mocked<KafkaApi>;
 
   beforeAll(async () => {
-    kafkaApi = {
+    devKafkaApi = {
       fetchTopicOffsets: jest.fn(),
       fetchGroupOffsets: jest.fn(),
     };
 
-    const router = makeRouter(getVoidLogger(), kafkaApi);
+    prodKafkaApi = {
+      fetchTopicOffsets: jest.fn(),
+      fetchGroupOffsets: jest.fn(),
+    };
+
+    apis = [
+      { name: 'dev', api: devKafkaApi },
+      { name: 'prod', api: prodKafkaApi },
+    ];
+
+    const router = makeRouter(getVoidLogger(), apis);
     app = express().use(router);
   });
 
@@ -39,7 +51,7 @@ describe('router', () => {
     jest.resetAllMocks();
   });
 
-  describe('get /consumer/:consumerId/offsets', () => {
+  describe('get /consumers/clusterId/:consumerId/offsets', () => {
     it('returns topic and group offsets', async () => {
       const topic1Offsets = [
         { id: 1, offset: '500' },
@@ -60,15 +72,17 @@ describe('router', () => {
           partitions: [{ id: 1, offset: '456' }],
         },
       ];
-      when(kafkaApi.fetchTopicOffsets)
+      when(prodKafkaApi.fetchTopicOffsets)
         .calledWith('topic1')
         .mockResolvedValue(topic1Offsets);
-      when(kafkaApi.fetchTopicOffsets)
+      when(prodKafkaApi.fetchTopicOffsets)
         .calledWith('topic2')
         .mockResolvedValue(topic2Offsets);
-      kafkaApi.fetchGroupOffsets.mockResolvedValue(groupOffsets);
+      when(prodKafkaApi.fetchGroupOffsets)
+        .calledWith('hey')
+        .mockResolvedValue(groupOffsets);
 
-      const response = await request(app).get('/consumer/hey/offsets');
+      const response = await request(app).get('/consumers/prod/hey/offsets');
 
       expect(response.status).toEqual(200);
       expect(response.body.consumerId).toEqual('hey');
@@ -98,11 +112,70 @@ describe('router', () => {
     });
 
     it('handles internal error correctly', async () => {
-      kafkaApi.fetchGroupOffsets.mockRejectedValue(Error('oh no'));
+      prodKafkaApi.fetchGroupOffsets.mockRejectedValue(Error('oh no'));
 
-      const response = await request(app).get('/consumer/hey/offsets');
+      const response = await request(app).get('/consumers/prod/hey/offsets');
 
       expect(response.status).toEqual(500);
     });
+
+    it('uses correct kafka cluster', async () => {
+      const topic1ProdOffsets = [{ id: 1, offset: '500' }];
+      const topic1DevOffsets = [{ id: 1, offset: '1234' }];
+      const groupProdOffsets = [
+        {
+          topic: 'topic1',
+          partitions: [{ id: 1, offset: '100' }],
+        },
+      ];
+      const groupDevOffsets = [
+        {
+          topic: 'topic1',
+          partitions: [{ id: 1, offset: '567' }],
+        },
+      ];
+      when(prodKafkaApi.fetchTopicOffsets)
+        .calledWith('topic1')
+        .mockResolvedValue(topic1ProdOffsets);
+      when(prodKafkaApi.fetchGroupOffsets)
+        .calledWith('hey')
+        .mockResolvedValue(groupProdOffsets);
+      when(devKafkaApi.fetchTopicOffsets)
+        .calledWith('topic1')
+        .mockResolvedValue(topic1DevOffsets);
+      when(devKafkaApi.fetchGroupOffsets)
+        .calledWith('hey')
+        .mockResolvedValue(groupDevOffsets);
+
+      const prodResponse = await request(app).get(
+        '/consumers/prod/hey/offsets',
+      );
+      const devResponse = await request(app).get('/consumers/dev/hey/offsets');
+
+      expect(prodResponse.status).toEqual(200);
+      expect(prodResponse.body.consumerId).toEqual('hey');
+      expect(new Set(prodResponse.body.offsets)).toStrictEqual(
+        new Set([
+          {
+            topic: 'topic1',
+            partitionId: 1,
+            groupOffset: '100',
+            topicOffset: '500',
+          },
+        ]),
+      );
+      expect(devResponse.status).toEqual(200);
+      expect(devResponse.body.consumerId).toEqual('hey');
+      expect(new Set(devResponse.body.offsets)).toStrictEqual(
+        new Set([
+          {
+            topic: 'topic1',
+            partitionId: 1,
+            groupOffset: '567',
+            topicOffset: '1234',
+          },
+        ]),
+      );
+    });
   });
 });

plugins/kafka-backend/src/service/router.ts

@@ -20,31 +20,43 @@ import { Logger } from 'winston';
 import { Config } from '@backstage/config';
 import { KafkaApi, KafkaJsApiImpl } from './KafkaApi';
 import _ from 'lodash';
-import { ConnectionOptions } from 'tls';
+import { getClusterDetails } from '../config/ClusterReader';
 
 export interface RouterOptions {
   logger: Logger;
   config: Config;
 }
 
+export interface ClusterApi {
+  name: string;
+  api: KafkaApi;
+}
+
 export const makeRouter = (
   logger: Logger,
-  kafkaApi: KafkaApi,
+  kafkaApis: ClusterApi[],
 ): express.Router => {
   const router = Router();
   router.use(express.json());
 
-  router.get('/consumer/:consumerId/offsets', async (req, res) => {
+  const kafkaApiByClusterName = _.keyBy(kafkaApis, item => item.name);
+
+  router.get('/consumers/:clusterId/:consumerId/offsets', async (req, res) => {
+    const clusterId = req.params.clusterId;
     const consumerId = req.params.consumerId;
 
-    logger.debug(`Fetch consumer group ${consumerId} offsets`);
+    logger.info(
+      `Fetch consumer group ${consumerId} offsets from cluster ${clusterId}`,
+    );
 
-    const groupOffsets = await kafkaApi.fetchGroupOffsets(consumerId);
+    const kafkaApi = kafkaApiByClusterName[clusterId];
+
+    const groupOffsets = await kafkaApi.api.fetchGroupOffsets(consumerId);
 
     const groupWithTopicOffsets = await Promise.all(
       groupOffsets.map(async ({ topic, partitions }) => {
         const topicOffsets = _.keyBy(
-          await kafkaApi.fetchTopicOffsets(topic),
+          await kafkaApi.api.fetchTopicOffsets(topic),
           partition => partition.id,
         );
 
@@ -70,12 +82,15 @@ export async function createRouter(
   logger.info('Initializing Kafka backend');
 
   const clientId = options.config.getString('kafka.clientId');
-  const brokers = options.config.getStringArray('kafka.brokers');
 
-  const sslConfig = options.config.getOptional('kafka.ssl');
-  const ssl = sslConfig ? (sslConfig as ConnectionOptions) : undefined;
+  const clusters = getClusterDetails(
+    options.config.getConfigArray('kafka.clusters'),
+  );
 
-  const kafkaApi = new KafkaJsApiImpl({ clientId, brokers, logger, ssl });
+  const kafkaApis = clusters.map(cluster => ({
+    name: cluster.name,
+    api: new KafkaJsApiImpl({ clientId, logger, ...cluster }),
+  }));
 
-  return makeRouter(logger, kafkaApi);
+  return makeRouter(logger, kafkaApis);
 }

plugins/kafka-backend/src/types/types.ts

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2020 Spotify AB
+ *
+ * Licensed 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.
+ */
+
+import { ConnectionOptions } from 'tls';
+
+export interface ClusterDetails {
+  name: string;
+  brokers: string[];
+  ssl?: ConnectionOptions;
+}

plugins/kafka/README.md

@@ -65,8 +65,10 @@ import { Router as KafkaRouter } from '@backstage/plugin-kafka';
 ```yaml
 kafka:
   clientId: backstage
-  brokers:
-    - localhost:9092
+  clusters:
+    - name: cluster-name
+      brokers:
+        - localhost:9092
 ```
 
 6. Add `kafka.apache.org/consumer-groups` annotation to your services:
@@ -77,7 +79,7 @@ kind: Component
 metadata:
   # ...
   annotations:
-    kafka.apache.org/consumer-groups: consumer-group-name
+    kafka.apache.org/consumer-groups: cluster-name/consumer-group-name
 spec:
   type: service
 ```

plugins/kafka/src/api/KafkaBackendClient.ts

@@ -43,8 +43,11 @@ export class KafkaBackendClient implements KafkaApi {
   }
 
   async getConsumerGroupOffsets(
+    clusterId: string,
     consumerGroup: string,
   ): Promise<ConsumerGroupOffsetsResponse> {
-    return await this.internalGet(`/consumer/${consumerGroup}/offsets`);
+    return await this.internalGet(
+      `/consumers/${clusterId}/${consumerGroup}/offsets`,
+    );
   }
 }

plugins/kafka/src/api/types.ts

@@ -34,6 +34,7 @@ export type ConsumerGroupOffsetsResponse = {
 
 export interface KafkaApi {
   getConsumerGroupOffsets(
+    clusterId: string,
     consumerGroup: string,
   ): Promise<ConsumerGroupOffsetsResponse>;
 }

plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.test.tsx

@@ -27,6 +27,7 @@ describe('ConsumerGroupOffsets', () => {
     const rendered = await renderInTestApp(
       <ConsumerGroupOffsets
         consumerGroup={consumerGroupOffsets.consumerId}
+        clusterId="prod"
         topics={consumerGroupOffsets.offsets}
         loading={false}
         retry={() => {}}

plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.tsx

@@ -15,7 +15,7 @@
  */
 
 import { Table, TableColumn } from '@backstage/core';
-import { Box, Typography } from '@material-ui/core';
+import { Box, Grid, Typography } from '@material-ui/core';
 import RetryIcon from '@material-ui/icons/Replay';
 import React from 'react';
 import { useConsumerGroupsOffsetsForEntity } from './useConsumerGroupsOffsetsForEntity';
@@ -62,6 +62,7 @@ const generatedColumns: TableColumn[] = [
 type Props = {
   loading: boolean;
   retry: () => void;
+  clusterId: string;
   consumerGroup: string;
   topics?: TopicPartitionInfo[];
 };
@@ -69,6 +70,7 @@ type Props = {
 export const ConsumerGroupOffsets = ({
   loading,
   topics,
+  clusterId,
   consumerGroup,
   retry,
 }: Props) => {
@@ -87,7 +89,7 @@ export const ConsumerGroupOffsets = ({
       title={
         <Box display="flex" alignItems="center">
           <Typography variant="h6">
-            Consumed Topics for {consumerGroup}
+            Consumed Topics for {consumerGroup} ({clusterId})
           </Typography>
         </Box>
       }
@@ -98,6 +100,15 @@ export const ConsumerGroupOffsets = ({
 
 export const KafkaTopicsForConsumer = () => {
   const [tableProps, { retry }] = useConsumerGroupsOffsetsForEntity();
-
-  return <ConsumerGroupOffsets {...tableProps} retry={retry} />;
+  return (
+    <Grid>
+      {tableProps.consumerGroupsTopics?.map(consumerGroup => (
+        <ConsumerGroupOffsets
+          {...consumerGroup}
+          loading={tableProps.loading}
+          retry={retry}
+        />
+      ))}
+    </Grid>
+  );
 };

plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.test.tsx

@@ -20,21 +20,7 @@ import { EntityContext } from '@backstage/plugin-catalog';
 import { Entity } from '@backstage/catalog-model';
 
 describe('useConsumerGroupOffsets', () => {
-  const entity: Entity = {
-    apiVersion: 'v1',
-    kind: 'Component',
-    metadata: {
-      name: 'test',
-      annotations: {
-        'kafka.apache.org/consumer-groups': 'consumer',
-      },
-    },
-    spec: {
-      owner: 'guest',
-      type: 'Website',
-      lifecycle: 'development',
-    },
-  };
+  let entity: Entity;
 
   const wrapper = ({ children }: PropsWithChildren<{}>) => {
     return (
@@ -46,8 +32,107 @@ describe('useConsumerGroupOffsets', () => {
 
   const subject = () => renderHook(useConsumerGroupsForEntity, { wrapper });
 
-  it('returns correct consumer group for annotation', async () => {
+  it('returns correct cluster and consumer group for annotation', async () => {
+    entity = {
+      apiVersion: 'v1',
+      kind: 'Component',
+      metadata: {
+        name: 'test',
+        annotations: {
+          'kafka.apache.org/consumer-groups': 'prod/consumer',
+        },
+      },
+      spec: {
+        owner: 'guest',
+        type: 'Website',
+        lifecycle: 'development',
+      },
+    };
     const { result } = subject();
-    expect(result.current).toBe('consumer');
+    expect(result.current).toStrictEqual([
+      {
+        clusterId: 'prod',
+        consumerGroup: 'consumer',
+      },
+    ]);
+  });
+
+  it('returns correct cluster and consumer group for multiple consumers', async () => {
+    entity = {
+      apiVersion: 'v1',
+      kind: 'Component',
+      metadata: {
+        name: 'test',
+        annotations: {
+          'kafka.apache.org/consumer-groups':
+            'prod/consumer,dev/another-consumer',
+        },
+      },
+      spec: {
+        owner: 'guest',
+        type: 'Website',
+        lifecycle: 'development',
+      },
+    };
+    const { result } = subject();
+    expect(result.current).toStrictEqual([
+      { clusterId: 'prod', consumerGroup: 'consumer' },
+      {
+        clusterId: 'dev',
+        consumerGroup: 'another-consumer',
+      },
+    ]);
+  });
+
+  it('returns correct cluster and consumer group for annotation with extra spaces', async () => {
+    entity = {
+      apiVersion: 'v1',
+      kind: 'Component',
+      metadata: {
+        name: 'test',
+        annotations: {
+          'kafka.apache.org/consumer-groups':
+            '   prod/consumer   ,   dev/another-consumer   ',
+        },
+      },
+      spec: {
+        owner: 'guest',
+        type: 'Website',
+        lifecycle: 'development',
+      },
+    };
+    const { result } = subject();
+    expect(result.current).toStrictEqual([
+      { clusterId: 'prod', consumerGroup: 'consumer' },
+      {
+        clusterId: 'dev',
+        consumerGroup: 'another-consumer',
+      },
+    ]);
+  });
+
+  it('fails on missing cluster', async () => {
+    entity = {
+      apiVersion: 'v1',
+      kind: 'Component',
+      metadata: {
+        name: 'test',
+        annotations: {
+          'kafka.apache.org/consumer-groups': 'dev/another,consumer',
+        },
+      },
+      spec: {
+        owner: 'guest',
+        type: 'Website',
+        lifecycle: 'development',
+      },
+    };
+    const { result } = subject();
+    expect(() => result.current).toThrowError();
+    expect(result.error).toStrictEqual(
+      new Error(
+        `Failed to parse kafka consumer group annotation: got "dev/another,consumer"`,
+      ),
+    );
   });
 });

plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.ts

@@ -15,10 +15,29 @@
  */
 
 import { useEntity } from '@backstage/plugin-catalog';
+import { useMemo } from 'react';
 import { KAFKA_CONSUMER_GROUP_ANNOTATION } from '../../constants';
 
 export const useConsumerGroupsForEntity = () => {
   const { entity } = useEntity();
+  const annotation =
+    entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? '';
 
-  return entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? '';
+  const consumerList = useMemo(() => {
+    return annotation.split(',').map(consumer => {
+      const [clusterId, consumerGroup] = consumer.split('/');
+
+      if (!clusterId || !consumerGroup) {
+        throw new Error(
+          `Failed to parse kafka consumer group annotation: got "${annotation}"`,
+        );
+      }
+      return {
+        clusterId: clusterId.trim(),
+        consumerGroup: consumerGroup.trim(),
+      };
+    });
+  }, [annotation]);
+
+  return consumerList;
 };

plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.test.tsx

@@ -45,7 +45,7 @@ describe('useConsumerGroupOffsets', () => {
     metadata: {
       name: 'test',
       annotations: {
-        'kafka.apache.org/consumer-groups': consumerGroupOffsets.consumerId,
+        'kafka.apache.org/consumer-groups': `prod/${consumerGroupOffsets.consumerId}`,
       },
     },
     spec: {
@@ -74,16 +74,24 @@ describe('useConsumerGroupOffsets', () => {
     renderHook(useConsumerGroupsOffsetsForEntity, { wrapper });
 
   it('returns correct consumer group for annotation', async () => {
+    mockKafkaApi.getConsumerGroupOffsets.mockResolvedValue(
+      consumerGroupOffsets,
+    );
     when(mockKafkaApi.getConsumerGroupOffsets)
-      .calledWith(consumerGroupOffsets.consumerId)
+      .calledWith('prod', consumerGroupOffsets.consumerId)
       .mockResolvedValue(consumerGroupOffsets);
 
     const { result, waitForNextUpdate } = subject();
     await waitForNextUpdate();
     const [tableProps] = result.current;
 
-    expect(tableProps.consumerGroup).toBe(consumerGroupOffsets.consumerId);
-    expect(tableProps.topics).toBe(consumerGroupOffsets.offsets);
+    expect(tableProps.consumerGroupsTopics).toStrictEqual([
+      {
+        clusterId: 'prod',
+        consumerGroup: consumerGroupOffsets.consumerId,
+        topics: consumerGroupOffsets.offsets,
+      },
+    ]);
   });
 
   it('posts an error to the error api', async () => {

plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.ts

@@ -20,25 +20,35 @@ import { kafkaApiRef } from '../../api/types';
 import { useConsumerGroupsForEntity } from './useConsumerGroupsForEntity';
 
 export const useConsumerGroupsOffsetsForEntity = () => {
-  const consumerGroup = useConsumerGroupsForEntity();
+  const consumers = useConsumerGroupsForEntity();
   const api = useApi(kafkaApiRef);
   const errorApi = useApi(errorApiRef);
 
-  const { loading, value: topics, retry } = useAsyncRetry(async () => {
+  const {
+    loading,
+    value: consumerGroupsTopics,
+    retry,
+  } = useAsyncRetry(async () => {
     try {
-      const response = await api.getConsumerGroupOffsets(consumerGroup);
-      return response.offsets;
+      return await Promise.all(
+        consumers.map(async ({ clusterId, consumerGroup }) => {
+          const response = await api.getConsumerGroupOffsets(
+            clusterId,
+            consumerGroup,
+          );
+          return { clusterId, consumerGroup, topics: response.offsets };
+        }),
+      );
     } catch (e) {
       errorApi.post(e);
       throw e;
     }
-  }, [api, errorApi, consumerGroup]);
+  }, [consumers, api, errorApi]);
 
   return [
     {
       loading,
-      consumerGroup,
-      topics,
+      consumerGroupsTopics,
     },
     {
       retry,