Added support for multiple kafka clusters

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.getRequired(`/consumer/${consumerGroup}/offsets`);
+    return await this.getRequired(
+      `/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/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,51 @@ 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('fails on missing cluster', async () => {
+    entity = {
+      apiVersion: 'v1',
+      kind: 'Component',
+      metadata: {
+        name: 'test',
+        annotations: {
+          'kafka.apache.org/consumer-groups': '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 "consumer"`,
+      ),
+    );
   });
 });

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

@@ -19,6 +19,15 @@ import { KAFKA_CONSUMER_GROUP_ANNOTATION } from '../../constants';
 
 export const useConsumerGroupsForEntity = () => {
   const { entity } = useEntity();
+  const annotation =
+    entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? '';
+  const [clusterId, consumerGroup] = annotation.split('/');
 
-  return entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? '';
+  if (!clusterId || !consumerGroup) {
+    throw new Error(
+      `Failed to parse kafka consumer group annotation: got "${annotation}"`,
+    );
+  }
+
+  return { clusterId, consumerGroup };
 };

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': `cluster/${consumerGroupOffsets.consumerId}`,
       },
     },
     spec: {
@@ -75,7 +75,7 @@ describe('useConsumerGroupOffsets', () => {
 
   it('returns correct consumer group for annotation', async () => {
     when(mockKafkaApi.getConsumerGroupOffsets)
-      .calledWith(consumerGroupOffsets.consumerId)
+      .calledWith('cluster', consumerGroupOffsets.consumerId)
       .mockResolvedValue(consumerGroupOffsets);
 
     const { result, waitForNextUpdate } = subject();

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

@@ -20,13 +20,16 @@ import { kafkaApiRef } from '../../api/types';
 import { useConsumerGroupsForEntity } from './useConsumerGroupsForEntity';
 
 export const useConsumerGroupsOffsetsForEntity = () => {
-  const consumerGroup = useConsumerGroupsForEntity();
+  const { clusterId, consumerGroup } = useConsumerGroupsForEntity();
   const api = useApi(kafkaApiRef);
   const errorApi = useApi(errorApiRef);
 
   const { loading, value: topics, retry } = useAsyncRetry(async () => {
     try {
-      const response = await api.getConsumerGroupOffsets(consumerGroup);
+      const response = await api.getConsumerGroupOffsets(
+        clusterId,
+        consumerGroup,
+      );
       return response.offsets;
     } catch (e) {
       errorApi.post(e);