Merge pull request #3985 from nirga/master

Initial version of new Kafka plugin
2021-01-20 19:22:22 +01:00
parent fc7fde42ff 5edf0ad930
commit 4c75de0610
36 changed files with 1478 additions and 4 deletions
@@ -0,0 +1,3 @@
+module.exports = {
+  extends: [require.resolve('@backstage/cli/config/eslint.backend')],
+};
@@ -0,0 +1,28 @@
+# Kafka Backend
+
+This is the backend part of the Kafka plugin. It responds to Kafka requests from the frontend.
+
+## Configuration
+
+This configures how to connect to the brokers in your Kafka cluster.
+
+### clientId
+
+The name of the client to use when connecting to the cluster.
+
+### brokers
+
+A list of the brokers' host names and ports to connect to.
+
+### SSL (optional)
+
+Configure TLS connection to the Kafka cluster. The options are passed directly to [tls.connect] and used to create the TLS secure context. Normally these would include `key` and `cert`.
+
+Example:
+
+```yaml
+kafka:
+  clientId: backstage
+  brokers:
+    - localhost:9092
+```
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+export interface Config {
+  kafka?: {
+    /**
+     * 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;
+    };
+  };
+}
@@ -0,0 +1,55 @@
+{
+  "name": "@backstage/plugin-kafka-backend",
+  "version": "0.1.0",
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "license": "Apache-2.0",
+  "private": false,
+  "publishConfig": {
+    "access": "public",
+    "main": "dist/index.cjs.js",
+    "types": "dist/index.d.ts"
+  },
+  "homepage": "https://backstage.io",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/backstage/backstage",
+    "directory": "plugins/kafka-backend"
+  },
+  "keywords": [
+    "backstage",
+    "kafka"
+  ],
+  "configSchema": "config.d.ts",
+  "scripts": {
+    "start": "backstage-cli backend:dev",
+    "build": "backstage-cli backend:build",
+    "lint": "backstage-cli lint",
+    "test": "backstage-cli test",
+    "prepack": "backstage-cli prepack",
+    "postpack": "backstage-cli postpack",
+    "clean": "backstage-cli clean"
+  },
+  "dependencies": {
+    "@backstage/backend-common": "^0.4.1",
+    "@backstage/catalog-model": "^0.6.0",
+    "@backstage/config": "^0.1.2",
+    "@types/express": "^4.17.6",
+    "express": "^4.17.1",
+    "express-promise-router": "^3.0.3",
+    "kafkajs": "^1.16.0-beta.6",
+    "lodash": "^4.17.15",
+    "winston": "^3.2.1"
+  },
+  "devDependencies": {
+    "@backstage/cli": "^0.4.3",
+    "@types/jest-when": "^2.7.2",
+    "@types/lodash": "^4.14.151",
+    "jest-when": "^3.1.0",
+    "supertest": "^4.0.2"
+  },
+  "files": [
+    "dist",
+    "schema.d.ts"
+  ]
+}
@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+
+export { createRouter } from './service/router';
@@ -0,0 +1,97 @@
+/*
+ * 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 { Kafka, SeekEntry } from 'kafkajs';
+import { Logger } from 'winston';
+import { ConnectionOptions } from 'tls';
+
+export type PartitionOffset = {
+  id: number;
+  offset: string;
+};
+
+export type TopicOffset = {
+  topic: string;
+  partitions: PartitionOffset[];
+};
+
+export type Options = {
+  clientId: string;
+  brokers: string[];
+  ssl?: ConnectionOptions;
+  logger: Logger;
+};
+
+export interface KafkaApi {
+  fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>>;
+  fetchGroupOffsets(groupId: string): Promise<Array<TopicOffset>>;
+}
+
+export class KafkaJsApiImpl implements KafkaApi {
+  private readonly kafka: Kafka;
+  private readonly logger: Logger;
+
+  constructor(options: Options) {
+    options.logger.debug(
+      `creating kafka client with clientId=${options.clientId} and brokers=${options.brokers}`,
+    );
+
+    this.kafka = new Kafka(options);
+    this.logger = options.logger;
+  }
+
+  async fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>> {
+    this.logger.debug(`fetching topic offsets for ${topic}`);
+
+    const admin = this.kafka.admin();
+    await admin.connect();
+
+    try {
+      return KafkaJsApiImpl.toPartitionOffsets(
+        await admin.fetchTopicOffsets(topic),
+      );
+    } finally {
+      await admin.disconnect();
+    }
+  }
+
+  async fetchGroupOffsets(groupId: string): Promise<Array<TopicOffset>> {
+    this.logger.debug(`fetching consumer group offsets for ${groupId}`);
+
+    const admin = this.kafka.admin();
+    await admin.connect();
+
+    try {
+      const groupOffsets = await admin.fetchOffsets({ groupId });
+
+      return groupOffsets.map(topicOffset => ({
+        topic: topicOffset.topic,
+        partitions: KafkaJsApiImpl.toPartitionOffsets(topicOffset.partitions),
+      }));
+    } finally {
+      await admin.disconnect();
+    }
+  }
+
+  private static toPartitionOffsets(
+    result: Array<SeekEntry>,
+  ): Array<PartitionOffset> {
+    return result.map(seekEntry => ({
+      id: seekEntry.partition,
+      offset: seekEntry.offset,
+    }));
+  }
+}
@@ -0,0 +1,108 @@
+/*
+ * 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 request from 'supertest';
+import express from 'express';
+import { makeRouter } 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>;
+
+  beforeAll(async () => {
+    kafkaApi = {
+      fetchTopicOffsets: jest.fn(),
+      fetchGroupOffsets: jest.fn(),
+    };
+
+    const router = makeRouter(getVoidLogger(), kafkaApi);
+    app = express().use(router);
+  });
+
+  beforeEach(() => {
+    jest.resetAllMocks();
+  });
+
+  describe('get /consumer/:consumerId/offsets', () => {
+    it('returns topic and group offsets', async () => {
+      const topic1Offsets = [
+        { id: 1, offset: '500' },
+        { id: 2, offset: '1000' },
+      ];
+      const topic2Offsets = [{ id: 1, offset: '456' }];
+
+      const groupOffsets = [
+        {
+          topic: 'topic1',
+          partitions: [
+            { id: 1, offset: '100' },
+            { id: 2, offset: '213' },
+          ],
+        },
+        {
+          topic: 'topic2',
+          partitions: [{ id: 1, offset: '456' }],
+        },
+      ];
+      when(kafkaApi.fetchTopicOffsets)
+        .calledWith('topic1')
+        .mockResolvedValue(topic1Offsets);
+      when(kafkaApi.fetchTopicOffsets)
+        .calledWith('topic2')
+        .mockResolvedValue(topic2Offsets);
+      kafkaApi.fetchGroupOffsets.mockResolvedValue(groupOffsets);
+
+      const response = await request(app).get('/consumer/hey/offsets');
+
+      expect(response.status).toEqual(200);
+      expect(response.body.consumerId).toEqual('hey');
+      // Note the Set comparison here since there's no guarantee on the order of the elements in the list.
+      expect(new Set(response.body.offsets)).toStrictEqual(
+        new Set([
+          {
+            topic: 'topic1',
+            partitionId: 1,
+            groupOffset: '100',
+            topicOffset: '500',
+          },
+          {
+            topic: 'topic1',
+            partitionId: 2,
+            groupOffset: '213',
+            topicOffset: '1000',
+          },
+          {
+            topic: 'topic2',
+            partitionId: 1,
+            groupOffset: '456',
+            topicOffset: '456',
+          },
+        ]),
+      );
+    });
+
+    it('handles internal error correctly', async () => {
+      kafkaApi.fetchGroupOffsets.mockRejectedValue(Error('oh no'));
+
+      const response = await request(app).get('/consumer/hey/offsets');
+
+      expect(response.status).toEqual(500);
+    });
+  });
+});
@@ -0,0 +1,81 @@
+/*
+ * 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 express from 'express';
+import Router from 'express-promise-router';
+import { Logger } from 'winston';
+import { Config } from '@backstage/config';
+import { KafkaApi, KafkaJsApiImpl } from './KafkaApi';
+import _ from 'lodash';
+import { ConnectionOptions } from 'tls';
+
+export interface RouterOptions {
+  logger: Logger;
+  config: Config;
+}
+
+export const makeRouter = (
+  logger: Logger,
+  kafkaApi: KafkaApi,
+): express.Router => {
+  const router = Router();
+  router.use(express.json());
+
+  router.get('/consumer/:consumerId/offsets', async (req, res) => {
+    const consumerId = req.params.consumerId;
+
+    logger.debug(`Fetch consumer group ${consumerId} offsets`);
+
+    const groupOffsets = await kafkaApi.fetchGroupOffsets(consumerId);
+
+    const groupWithTopicOffsets = await Promise.all(
+      groupOffsets.map(async ({ topic, partitions }) => {
+        const topicOffsets = _.keyBy(
+          await kafkaApi.fetchTopicOffsets(topic),
+          partition => partition.id,
+        );
+
+        return partitions.map(partition => ({
+          topic: topic,
+          partitionId: partition.id,
+          groupOffset: partition.offset,
+          topicOffset: topicOffsets[partition.id].offset,
+        }));
+      }),
+    );
+    res.send({ consumerId, offsets: groupWithTopicOffsets.flat() });
+  });
+
+  return router;
+};
+
+export async function createRouter(
+  options: RouterOptions,
+): Promise<express.Router> {
+  const logger = options.logger;
+
+  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 kafkaApi = new KafkaJsApiImpl({ clientId, brokers, logger, ssl });
+
+  return makeRouter(logger, kafkaApi);
+}
@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+
+export {};
@@ -0,0 +1,3 @@
+module.exports = {
+  extends: [require.resolve('@backstage/cli/config/eslint')],
+};
@@ -0,0 +1,87 @@
+# Kafka Plugin
+
+<img src="./src/assets/screenshot-1.png">
+
+## Setup
+
+1. Run:
+
+```bash
+yarn add @backstage/plugin-kafka @backstage/plugin-kafka-backend
+```
+
+2. Add the plugin backend:
+
+In a new file named `kafka.js` under `backend/src/plugins`:
+
+```js
+import { createRouter } from '@backstage/plugin-kafka-backend';
+import { PluginEnvironment } from '../types';
+
+export default async function createPlugin({
+  logger,
+  config,
+}: PluginEnvironment) {
+  return await createRouter({ logger, config });
+}
+```
+
+And then add to `packages/backend/src/index.ts`:
+
+```js
+async function main() {
+  // ...
+  const kafkaEnv = useHotMemoize(module, () => createEnv('kafka'));
+  // ...
+
+  const apiRouter = Router();
+  // ...
+  apiRouter.use('/kafka', await kafka(kafkaEnv));
+  // ...
+```
+
+3. Add the plugin frontend to `packages/app/src/plugin.ts`:
+
+```js
+export { plugin as Kafka } from '@backstage/plugin-kafka';
+```
+
+4. Register the plugin frontend router in `packages/app/src/components/catalog/EntityPage.tsx`:
+
+```jsx
+import { Router as KafkaRouter } from '@backstage/plugin-kafka';
+
+// Then, somewhere inside <EntityPageLayout>
+
+<EntityPageLayout.Content
+  path="/kafka/*"
+  title="Kafka"
+  element={<KafkaRouter entity={entity} />}
+/>;
+```
+
+5. Add broker configs for the backend in your `app-config.yaml`:
+
+```yaml
+kafka:
+  clientId: backstage
+  brokers:
+    - localhost:9092
+```
+
+6. Add `kafka.apache.org/consumer-groups` annotation to your services:
+
+```yaml
+apiVersion: backstage.io/v1alpha1
+kind: Component
+metadata:
+  # ...
+  annotations:
+    kafka.apache.org/consumer-groups: consumer-group-name
+spec:
+  type: service
+```
+
+## Features
+
+- List topics offsets and consumer group offsets for configured services.
@@ -0,0 +1,19 @@
+/*
+ * 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 { createDevApp } from '@backstage/dev-utils';
+import { plugin } from '../src/plugin';
+
+createDevApp().registerPlugin(plugin).render();
@@ -0,0 +1,52 @@
+{
+  "name": "@backstage/plugin-kafka",
+  "version": "0.1.0",
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "license": "Apache-2.0",
+  "publishConfig": {
+    "access": "public",
+    "main": "dist/index.esm.js",
+    "types": "dist/index.d.ts"
+  },
+  "scripts": {
+    "build": "backstage-cli plugin:build",
+    "start": "backstage-cli plugin:serve",
+    "lint": "backstage-cli lint",
+    "test": "backstage-cli test",
+    "diff": "backstage-cli plugin:diff",
+    "prepack": "backstage-cli prepack",
+    "postpack": "backstage-cli postpack",
+    "clean": "backstage-cli clean"
+  },
+  "dependencies": {
+    "@backstage/catalog-model": "^0.6.0",
+    "@backstage/core": "^0.4.4",
+    "@backstage/plugin-catalog": "^0.2.7",
+    "@backstage/theme": "^0.2.2",
+    "@material-ui/core": "^4.11.0",
+    "@material-ui/icons": "^4.9.1",
+    "@material-ui/lab": "4.0.0-alpha.45",
+    "react": "^16.13.1",
+    "react-dom": "^16.13.1",
+    "react-router": "6.0.0-beta.0",
+    "react-use": "^15.3.3"
+  },
+  "devDependencies": {
+    "@backstage/cli": "^0.4.6",
+    "@backstage/dev-utils": "^0.1.7",
+    "@backstage/test-utils": "^0.1.6",
+    "@testing-library/jest-dom": "^5.10.1",
+    "@testing-library/react": "^10.4.1",
+    "@testing-library/react-hooks": "^3.4.2",
+    "@testing-library/user-event": "^12.0.7",
+    "@types/jest": "^26.0.7",
+    "@types/node": "^12.0.0",
+    "cross-fetch": "^3.0.6",
+    "jest-when": "^3.1.0",
+    "msw": "^0.21.2"
+  },
+  "files": [
+    "dist"
+  ]
+}
@@ -0,0 +1,40 @@
+/*
+ * 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 { Entity } from '@backstage/catalog-model';
+import React from 'react';
+import { Route, Routes } from 'react-router';
+
+import { rootCatalogKafkaRouteRef } from './plugin';
+import { KAFKA_CONSUMER_GROUP_ANNOTATION } from './constants';
+import { KafkaTopicsForConsumer } from './components/ConsumerGroupOffsets/ConsumerGroupOffsets';
+import { MissingAnnotationEmptyState } from '@backstage/core';
+
+export const isPluginApplicableToEntity = (entity: Entity) =>
+  Boolean(entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION]);
+
+export const Router = ({ entity }: { entity: Entity }) => {
+  return !isPluginApplicableToEntity(entity) ? (
+    <MissingAnnotationEmptyState annotation={KAFKA_CONSUMER_GROUP_ANNOTATION} />
+  ) : (
+    <Routes>
+      <Route
+        path={`${rootCatalogKafkaRouteRef.path}`}
+        element={<KafkaTopicsForConsumer />}
+      />
+    </Routes>
+  );
+};
@@ -0,0 +1,50 @@
+/*
+ * 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 { DiscoveryApi } from '@backstage/core';
+import { KafkaApi, ConsumerGroupOffsetsResponse } from './types';
+
+export class KafkaBackendClient implements KafkaApi {
+  private readonly discoveryApi: DiscoveryApi;
+
+  constructor(options: { discoveryApi: DiscoveryApi }) {
+    this.discoveryApi = options.discoveryApi;
+  }
+
+  private async internalGet(path: string): Promise<any> {
+    const url = `${await this.discoveryApi.getBaseUrl('kafka')}${path}`;
+    const response = await fetch(url, {
+      method: 'GET',
+      headers: {
+        'Content-Type': 'application/json',
+      },
+    });
+
+    if (!response.ok) {
+      const payload = await response.text();
+      const message = `Request failed with ${response.status} ${response.statusText}, ${payload}`;
+      throw new Error(message);
+    }
+
+    return await response.json();
+  }
+
+  async getConsumerGroupOffsets(
+    consumerGroup: string,
+  ): Promise<ConsumerGroupOffsetsResponse> {
+    return await this.internalGet(`/consumer/${consumerGroup}/offsets`);
+  }
+}
@@ -0,0 +1,39 @@
+/*
+ * 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 { createApiRef } from '@backstage/core';
+
+export const kafkaApiRef = createApiRef<KafkaApi>({
+  id: 'plugin.kafka.service',
+  description:
+    'Used by the Kafka plugin to make requests to accompanying backend',
+});
+
+export type ConsumerGroupOffsetsResponse = {
+  consumerId: string;
+  offsets: {
+    topic: string;
+    partitionId: number;
+    topicOffset: string;
+    groupOffset: string;
+  }[];
+};
+
+export interface KafkaApi {
+  getConsumerGroupOffsets(
+    consumerGroup: string,
+  ): Promise<ConsumerGroupOffsetsResponse>;
+}
@@ -0,0 +1,40 @@
+/*
+ * 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 React from 'react';
+import { renderInTestApp } from '@backstage/test-utils';
+import { ConsumerGroupOffsets } from './ConsumerGroupOffsets';
+import * as data from './__fixtures__/consumer-group-offsets.json';
+import { ConsumerGroupOffsetsResponse } from '../../api/types';
+
+const consumerGroupOffsets = data as ConsumerGroupOffsetsResponse;
+
+describe('ConsumerGroupOffsets', () => {
+  it('should render consumer group table', async () => {
+    const rendered = await renderInTestApp(
+      <ConsumerGroupOffsets
+        consumerGroup={consumerGroupOffsets.consumerId}
+        topics={consumerGroupOffsets.offsets}
+        loading={false}
+        retry={() => {}}
+      />,
+    );
+
+    expect(rendered.getByText(/consumer/)).toBeInTheDocument();
+    expect(rendered.getByText('topic1')).toBeInTheDocument();
+    expect(rendered.getByText('topic2')).toBeInTheDocument();
+  });
+});
@@ -0,0 +1,103 @@
+/*
+ * 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 { Table, TableColumn } from '@backstage/core';
+import { Box, Typography } from '@material-ui/core';
+import RetryIcon from '@material-ui/icons/Replay';
+import React from 'react';
+import { useConsumerGroupsOffsetsForEntity } from './useConsumerGroupsOffsetsForEntity';
+
+export type TopicPartitionInfo = {
+  topic: string;
+  partitionId: number;
+  topicOffset: string;
+  groupOffset: string;
+};
+
+const generatedColumns: TableColumn[] = [
+  {
+    title: 'Topic',
+    field: 'topic',
+    highlight: true,
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.topic ?? ''}</>;
+    },
+  },
+  {
+    title: 'Partition',
+    field: 'partitionId',
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.partitionId ?? ''}</>;
+    },
+  },
+  {
+    title: 'Topic Offset',
+    field: 'topicOffset',
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.topicOffset ?? ''}</>;
+    },
+  },
+  {
+    title: 'Group Offset',
+    field: 'groupOffset',
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.groupOffset ?? ''}</>;
+    },
+  },
+];
+
+type Props = {
+  loading: boolean;
+  retry: () => void;
+  consumerGroup: string;
+  topics?: TopicPartitionInfo[];
+};
+
+export const ConsumerGroupOffsets = ({
+  loading,
+  topics,
+  consumerGroup,
+  retry,
+}: Props) => {
+  return (
+    <Table
+      isLoading={loading}
+      actions={[
+        {
+          icon: () => <RetryIcon />,
+          tooltip: 'Refresh Data',
+          isFreeAction: true,
+          onClick: () => retry(),
+        },
+      ]}
+      data={topics ?? []}
+      title={
+        <Box display="flex" alignItems="center">
+          <Typography variant="h6">
+            Consumed Topics for {consumerGroup}
+          </Typography>
+        </Box>
+      }
+      columns={generatedColumns}
+    />
+  );
+};
+
+export const KafkaTopicsForConsumer = () => {
+  const [tableProps, { retry }] = useConsumerGroupsOffsetsForEntity();
+
+  return <ConsumerGroupOffsets {...tableProps} retry={retry} />;
+};
@@ -0,0 +1,17 @@
+{
+  "consumerId": "consumer",
+  "offsets": [
+    {
+      "topic": "topic1",
+      "partitionId": 1,
+      "topicOffset": "100",
+      "groupOffset": "50"
+    },
+    {
+      "topic": "topic2",
+      "partitionId": 1,
+      "topicOffset": "2340",
+      "groupOffset": "1234"
+    }
+  ]
+}
@@ -0,0 +1,53 @@
+/*
+ * 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 React, { PropsWithChildren } from 'react';
+import { renderHook } from '@testing-library/react-hooks';
+import { useConsumerGroupsForEntity } from './useConsumerGroupsForEntity';
+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',
+    },
+  };
+
+  const wrapper = ({ children }: PropsWithChildren<{}>) => {
+    return (
+      <EntityContext.Provider value={{ entity: entity, loading: false }}>
+        {children}
+      </EntityContext.Provider>
+    );
+  };
+
+  const subject = () => renderHook(useConsumerGroupsForEntity, { wrapper });
+
+  it('returns correct consumer group for annotation', async () => {
+    const { result } = subject();
+    expect(result.current).toBe('consumer');
+  });
+});
@@ -0,0 +1,24 @@
+/*
+ * 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 { useEntity } from '@backstage/plugin-catalog';
+import { KAFKA_CONSUMER_GROUP_ANNOTATION } from '../../constants';
+
+export const useConsumerGroupsForEntity = () => {
+  const { entity } = useEntity();
+
+  return entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? '';
+};
@@ -0,0 +1,98 @@
+/*
+ * 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 React, { PropsWithChildren } from 'react';
+import { renderHook } from '@testing-library/react-hooks';
+import { EntityContext } from '@backstage/plugin-catalog';
+import { Entity } from '@backstage/catalog-model';
+import * as data from './__fixtures__/consumer-group-offsets.json';
+import {
+  ConsumerGroupOffsetsResponse,
+  KafkaApi,
+  kafkaApiRef,
+} from '../../api/types';
+import { ApiProvider, ApiRegistry, errorApiRef } from '@backstage/core';
+import { useConsumerGroupsOffsetsForEntity } from './useConsumerGroupsOffsetsForEntity';
+import { when } from 'jest-when';
+
+const consumerGroupOffsets = data as ConsumerGroupOffsetsResponse;
+
+const mockErrorApi: jest.Mocked<typeof errorApiRef.T> = {
+  post: jest.fn(),
+  error$: jest.fn(),
+};
+
+const mockKafkaApi: jest.Mocked<KafkaApi> = {
+  getConsumerGroupOffsets: jest.fn(),
+};
+
+describe('useConsumerGroupOffsets', () => {
+  const entity: Entity = {
+    apiVersion: 'v1',
+    kind: 'Component',
+    metadata: {
+      name: 'test',
+      annotations: {
+        'kafka.apache.org/consumer-groups': consumerGroupOffsets.consumerId,
+      },
+    },
+    spec: {
+      owner: 'guest',
+      type: 'Website',
+      lifecycle: 'development',
+    },
+  };
+
+  const wrapper = ({ children }: PropsWithChildren<{}>) => {
+    return (
+      <ApiProvider
+        apis={ApiRegistry.with(errorApiRef, mockErrorApi).with(
+          kafkaApiRef,
+          mockKafkaApi,
+        )}
+      >
+        <EntityContext.Provider value={{ entity: entity, loading: false }}>
+          {children}
+        </EntityContext.Provider>
+      </ApiProvider>
+    );
+  };
+
+  const subject = () =>
+    renderHook(useConsumerGroupsOffsetsForEntity, { wrapper });
+
+  it('returns correct consumer group for annotation', async () => {
+    when(mockKafkaApi.getConsumerGroupOffsets)
+      .calledWith(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);
+  });
+
+  it('posts an error to the error api', async () => {
+    const error = new Error('error!');
+    mockKafkaApi.getConsumerGroupOffsets.mockRejectedValueOnce(error);
+
+    const { waitForNextUpdate } = subject();
+    await waitForNextUpdate();
+
+    expect(mockErrorApi.post).toHaveBeenCalledWith(error);
+  });
+});
@@ -0,0 +1,47 @@
+/*
+ * 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 { errorApiRef, useApi } from '@backstage/core';
+import { useAsyncRetry } from 'react-use';
+import { kafkaApiRef } from '../../api/types';
+import { useConsumerGroupsForEntity } from './useConsumerGroupsForEntity';
+
+export const useConsumerGroupsOffsetsForEntity = () => {
+  const consumerGroup = useConsumerGroupsForEntity();
+  const api = useApi(kafkaApiRef);
+  const errorApi = useApi(errorApiRef);
+
+  const { loading, value: topics, retry } = useAsyncRetry(async () => {
+    try {
+      const response = await api.getConsumerGroupOffsets(consumerGroup);
+      return response.offsets;
+    } catch (e) {
+      errorApi.post(e);
+      throw e;
+    }
+  }, [api, errorApi, consumerGroup]);
+
+  return [
+    {
+      loading,
+      consumerGroup,
+      topics,
+    },
+    {
+      retry,
+    },
+  ] as const;
+};
@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+export const KAFKA_CONSUMER_GROUP_ANNOTATION =
+  'kafka.apache.org/consumer-groups';
@@ -0,0 +1,18 @@
+/*
+ * 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.
+ */
+export { plugin } from './plugin';
+export { KAFKA_CONSUMER_GROUP_ANNOTATION } from './constants';
+export { Router, isPluginApplicableToEntity } from './Router';
@@ -0,0 +1,22 @@
+/*
+ * 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 { plugin } from './plugin';
+
+describe('kafka', () => {
+  it('should export plugin', () => {
+    expect(plugin).toBeDefined();
+  });
+});
@@ -0,0 +1,39 @@
+/*
+ * 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 {
+  createApiFactory,
+  createPlugin,
+  createRouteRef,
+  discoveryApiRef,
+} from '@backstage/core';
+import { KafkaBackendClient } from './api/KafkaBackendClient';
+import { kafkaApiRef } from './api/types';
+
+export const rootCatalogKafkaRouteRef = createRouteRef({
+  path: '*',
+  title: 'Kafka',
+});
+
+export const plugin = createPlugin({
+  id: 'kafka',
+  apis: [
+    createApiFactory({
+      api: kafkaApiRef,
+      deps: { discoveryApi: discoveryApiRef },
+      factory: ({ discoveryApi }) => new KafkaBackendClient({ discoveryApi }),
+    }),
+  ],
+});
@@ -0,0 +1,17 @@
+/*
+ * 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 '@testing-library/jest-dom';
+import 'cross-fetch/polyfill';