From fe1fb4d60b465f71f16cc89a752f0619d3f399d5 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 19 Mar 2025 13:00:57 +0100 Subject: [PATCH 01/15] feat(events): add events kafka module Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/.eslintrc.js | 1 + plugins/events-backend-module-kafka/README.md | 78 +++++++ .../catalog-info.yaml | 10 + .../events-backend-module-kafka/config.d.ts | 202 ++++++++++++++++++ .../events-backend-module-kafka/package.json | 41 ++++ .../events-backend-module-kafka/report.api.md | 50 +++++ .../events-backend-module-kafka/src/index.ts | 19 ++ .../src/publisher/KafkaConsumerClient.test.ts | 118 ++++++++++ .../src/publisher/KafkaConsumerClient.ts | 67 ++++++ .../KafkaConsumingEventPublisher.test.ts | 92 ++++++++ .../publisher/KafkaConsumingEventPublisher.ts | 104 +++++++++ .../src/publisher/LoggerServiceCreator.ts | 45 ++++ .../src/publisher/config.test.ts | 197 +++++++++++++++++ .../src/publisher/config.ts | 90 ++++++++ ...ModuleKafkaConsumingEventPublisher.test.ts | 85 ++++++++ ...ventsModuleKafkaConsumingEventPublisher.ts | 52 +++++ yarn.lock | 7 + 17 files changed, 1258 insertions(+) create mode 100644 plugins/events-backend-module-kafka/.eslintrc.js create mode 100644 plugins/events-backend-module-kafka/README.md create mode 100644 plugins/events-backend-module-kafka/catalog-info.yaml create mode 100644 plugins/events-backend-module-kafka/config.d.ts create mode 100644 plugins/events-backend-module-kafka/package.json create mode 100644 plugins/events-backend-module-kafka/report.api.md create mode 100644 plugins/events-backend-module-kafka/src/index.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/config.test.ts create mode 100644 plugins/events-backend-module-kafka/src/publisher/config.ts create mode 100644 plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts create mode 100644 plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts diff --git a/plugins/events-backend-module-kafka/.eslintrc.js b/plugins/events-backend-module-kafka/.eslintrc.js new file mode 100644 index 0000000000..e2a53a6ad2 --- /dev/null +++ b/plugins/events-backend-module-kafka/.eslintrc.js @@ -0,0 +1 @@ +module.exports = require('@backstage/cli/config/eslint-factory')(__dirname); diff --git a/plugins/events-backend-module-kafka/README.md b/plugins/events-backend-module-kafka/README.md new file mode 100644 index 0000000000..0bdbd5cf1e --- /dev/null +++ b/plugins/events-backend-module-kafka/README.md @@ -0,0 +1,78 @@ +# @backstage/backstage-plugin-events-backend-module-kafka + +Welcome to the `events-backend-module-kafka` backend module! + +This package is a module for the `events-backend` backend plugin and extends the events system with an `KafkaConsumingEventPublisher.` + +This event publisher will allow you to receive events from an Kafka queue and will publish these to the used `EventsService` implementation. + +## Configuration + +To set up Kafka queues, you need to configure the following values: + +```yaml +events: + modules: + kafka: + kafkaConsumingEventPublisher: + clientId: your-client-id # (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. + brokers: # (Required) List of brokers in the Kafka cluster to connect to. + - broker1 + - broker2 + topics: + - topic: 'backstage.topic' # (Required) Replace with actual topic name as expected by subscribers + kafka: + topics: # (Required) The Kafka topics to subscribe to. + - topic1 + groupId: your-group-id # (Required) The GroupId to be used by the topic consumers. +``` + +For a complete list of all available fields that can be configured, refer to the [config.d.ts file](path/to/config.d.ts). + +### Optional SSL Configuration + +If your Kafka cluster requires SSL, you can configure it as follows: + +```yaml +events: + modules: + kafka: + kafkaConsumingEventPublisher: + ssl: + rejectUnauthorized: true # (Optional) If true, the server certificate is verified against the list of supplied CAs. + ca: [path/to/ca-cert] # (Optional) Array of trusted certificates in PEM format. + key: path/to/client-key # (Optional) Private key in PEM format. + cert: path/to/client-cert # (Optional) Public x509 certificate in PEM format. +``` + +### Optional SASL Authentication Configuration + +If your Kafka cluster requires `SASL` authentication, you can configure it as follows: + +```yaml +events: + modules: + kafka: + kafkaConsumingEventPublisher: + sasl: + mechanism: 'plain' # SASL mechanism ('plain', 'scram-sha-256' or 'scram-sha-512') + username: your-username # SASL username + password: your-password # SASL password +``` + +This section includes optional `SSL` and `SASL` authentication configuration for enhanced security. + +## Installation + +1. Install this module +2. Add your configuration. + +```bash +# From your Backstage root directory +yarn --cwd packages/backend add @backstage/plugin-events-backend-module-kafka +``` + +```typescript +// packages/backend/src/index.ts +backend.add(import('@backstage/plugin-events-backend-module-kafka')); +``` diff --git a/plugins/events-backend-module-kafka/catalog-info.yaml b/plugins/events-backend-module-kafka/catalog-info.yaml new file mode 100644 index 0000000000..72358ec3c3 --- /dev/null +++ b/plugins/events-backend-module-kafka/catalog-info.yaml @@ -0,0 +1,10 @@ +apiVersion: backstage.io/v1alpha1 +kind: Component +metadata: + name: backstage-plugin-events-backend-module-kafka + title: '@backstage/plugin-events-backend-module-kafka' + description: The kafka backend module for the events plugin. +spec: + lifecycle: experimental + type: backstage-backend-plugin-module + owner: maintainers diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts new file mode 100644 index 0000000000..4c0d06c77e --- /dev/null +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -0,0 +1,202 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { + events?: { + modules?: { + /** + * events-backend-module-kafka plugin configuration. + */ + kafka?: { + /** + * Configuration for KafkaConsumingEventPublisher + */ + kafkaConsumingEventPublisher?: { + /** + * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. + */ + clientId: string; + /** + * (Required) 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; + rejectUnauthorized?: boolean; + } + | boolean; + /** + * Optional SASL connection parameters. + */ + sasl?: { + mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512'; + username: string; + /** @visibility secret */ + password: string; + }; + + /** + * Optional retry connection parameters. + */ + retry: { + /** + * (Optional) Maximum wait time for a retry in milliseconds + * Default: 30000 ms. + */ + maxRetryTime: number; + + /** + * (Optional) Initial value used to calculate the retry in milliseconds (This is still randomized following the randomization factor) + * Default: 300 ms. + */ + initialRetryTime: number; + + /** + * (Optional) Randomization factor + * Default: 0.2. + */ + factor: number; + + /** + * (Optional) Exponential factor + * Default: 2. + */ + multiplier: number; + + /** + * (Optional) Max number of retries per call + * Default: 5. + */ + retries: number; + }; + + /** + * (Optional) Timeout in ms for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout: number; + + /** + * (Optional) Time in milliseconds to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout: number; + + /** + * (Optional) Time in milliseconds to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout: number; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout: boolean; + + /** + * Contains a object per topic for which an Kafka queue + * should be used as source of events. + */ + topics: Array<{ + /** + * (Required) The Backstage topic to publish to + */ + topic: string; + /** + * (Required) KafkaConsumer-related configuration. + */ + kafka: { + /** + * (Required) The Kafka topics to subscribe to + */ + topics: string[]; + /** + * (Required) The GroupId to be used by the topic consumers + */ + groupId: string; + + /** + * (Optional) Timeout in milliseconds used to detect failures. + * The consumer sends periodic heartbeats to indicate its liveness to the broker. + * If no heartbeats are received by the broker before the expiration of this session timeout, + * then the broker will remove this consumer from the group and initiate a rebalance + * Default: 30000 ms. + */ + sessionTimeout: number; + + /** + * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group + * Default: 60000 ms. + */ + rebalanceTimeout: number; + + /** + * (Optional) The expected time in milliseconds between heartbeats to the consumer coordinator. + * Heartbeats are used to ensure that the consumer's session stays active. + * The value must be set lower than session timeout + * Default: 3000 ms. + */ + heartbeatInterval: number; + + /** + * (Optional) The period of time in milliseconds after which we force a refresh of metadata + * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions + * Default: 300000 ms (5 minutes). + */ + metadataMaxAge: number; + + /** + * (Optional) The maximum amount of data per-partition the server will return. + * This size must be at least as large as the maximum message size the server allows + * or else it is possible for the producer to send messages larger than the consumer can fetch. + * If that happens, the consumer can get stuck trying to fetch a large message on a certain partition + * Default: 1048576 (1MB) + */ + maxBytesPerPartition: number; + + /** + * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTimeInMs for more data to accumulate. + * Default: 1 + */ + minBytes: number; + + /** + * (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0 + * Default: 10485760 (10MB) + */ + maxBytes: number; + + /** + * (Optional) The maximum amount of time in milliseconds the server will block before answering the fetch request + * if there isn’t sufficient data to immediately satisfy the requirement given by minBytes + * Default: 5000 + */ + maxWaitTimeInMs: number; + }; + }>; + }; + }; + }; + }; +} diff --git a/plugins/events-backend-module-kafka/package.json b/plugins/events-backend-module-kafka/package.json new file mode 100644 index 0000000000..1750caf36b --- /dev/null +++ b/plugins/events-backend-module-kafka/package.json @@ -0,0 +1,41 @@ +{ + "name": "@backstage/plugin-events-backend-module-kafka", + "description": "The kafka backend module for the events plugin.", + "version": "0.0.0", + "main": "src/index.ts", + "types": "src/index.ts", + "license": "Apache-2.0", + "private": true, + "publishConfig": { + "access": "public", + "main": "dist/index.cjs.js", + "types": "dist/index.d.ts" + }, + "backstage": { + "role": "backend-plugin-module" + }, + "scripts": { + "start": "backstage-cli package start", + "build": "backstage-cli package build", + "lint": "backstage-cli package lint", + "test": "backstage-cli package test", + "clean": "backstage-cli package clean", + "prepack": "backstage-cli package prepack", + "postpack": "backstage-cli package postpack" + }, + "dependencies": { + "@backstage/backend-plugin-api": "workspace:^", + "@backstage/config": "workspace:^", + "@backstage/plugin-events-node": "workspace:^", + "@backstage/types": "workspace:^", + "kafkajs": "^2.2.4" + }, + "devDependencies": { + "@backstage/backend-test-utils": "workspace:^", + "@backstage/cli": "workspace:^", + "@backstage/plugin-events-backend-test-utils": "workspace:^" + }, + "files": [ + "dist" + ] +} diff --git a/plugins/events-backend-module-kafka/report.api.md b/plugins/events-backend-module-kafka/report.api.md new file mode 100644 index 0000000000..b8fd142d0e --- /dev/null +++ b/plugins/events-backend-module-kafka/report.api.md @@ -0,0 +1,50 @@ +## API Report File for "@internal/plugin-events-backend-module-kafka" + +> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). + +```ts +import { BackendFeature } from '@backstage/backend-plugin-api'; +import { Config } from '@backstage/config'; +import { ConsumerConfig } from 'kafkajs'; +import { ConsumerSubscribeTopics } from 'kafkajs'; +import { EventsService } from '@backstage/plugin-events-node'; +import { Kafka } from 'kafkajs'; +import { LoggerService } from '@backstage/backend-plugin-api'; + +// @public +const eventsModuleKafkaConsumingEventPublisher: BackendFeature; +export default eventsModuleKafkaConsumingEventPublisher; + +// @public (undocumented) +export class KafkaConsumerClient { + // (undocumented) + static fromConfig(env: { + config: Config; + events: EventsService; + logger: LoggerService; + }): KafkaConsumerClient; + // (undocumented) + shutdown(): Promise; + // (undocumented) + start(): Promise; +} + +// @public (undocumented) +export class KafkaConsumingEventPublisher { + // (undocumented) + static fromConfig(env: { + kafkaClient: Kafka; + config: KafkaConsumerConfig; + events: EventsService; + logger: LoggerService; + }): KafkaConsumingEventPublisher; + // (undocumented) + shutdown(): Promise; + // (undocumented) + start(): Promise; +} + +// Warnings were encountered during analysis: +// +// src/publisher/KafkaConsumingEventPublisher.d.ts:13:9 - (ae-forgotten-export) The symbol "KafkaConsumerConfig" needs to be exported by the entry point index.d.ts +``` diff --git a/plugins/events-backend-module-kafka/src/index.ts b/plugins/events-backend-module-kafka/src/index.ts new file mode 100644 index 0000000000..ff4fe2a209 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/index.ts @@ -0,0 +1,19 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { KafkaConsumerClient } from './publisher/KafkaConsumerClient'; +export { KafkaConsumingEventPublisher } from './publisher/KafkaConsumingEventPublisher'; +export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher'; diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts new file mode 100644 index 0000000000..c0d6809514 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts @@ -0,0 +1,118 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { KafkaConsumerClient } from './KafkaConsumerClient'; +import { ConfigReader } from '@backstage/config'; +import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; +import { mockServices } from '@backstage/backend-test-utils'; + +jest.mock('kafkajs'); +jest.mock('./KafkaConsumingEventPublisher'); + +describe('KafkaConsumerClient', () => { + const mockLogger = mockServices.logger.mock(); + const mockEvents = mockServices.events.mock(); + + const mockConfig = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: [ + { + topic: 'fake1', + kafka: { + topics: ['topic-A'], + groupId: 'my-group', + }, + }, + { + topic: 'fake2', + kafka: { + topics: ['topic-B'], + groupId: 'my-group', + }, + }, + ], + }, + }, + }, + }, + }); + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('should create an instance from config', () => { + const client = KafkaConsumerClient.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(client).toBeInstanceOf(KafkaConsumerClient); + }); + + it('should create a consumer for each topic from config', () => { + KafkaConsumerClient.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(KafkaConsumingEventPublisher.fromConfig).toHaveBeenCalledTimes(2); + }); + + it('should start all consumers', async () => { + const mockConsumer = { + start: jest.fn().mockResolvedValue(undefined), + }; + (KafkaConsumingEventPublisher.fromConfig as jest.Mock).mockReturnValue( + mockConsumer, + ); + + const client = KafkaConsumerClient.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + await client.start(); + + expect(mockConsumer.start).toHaveBeenCalled(); + }); + + it('should shutdown all consumers', async () => { + const mockConsumer = { + shutdown: jest.fn().mockResolvedValue(undefined), + }; + (KafkaConsumingEventPublisher.fromConfig as jest.Mock).mockReturnValue( + mockConsumer, + ); + + const client = KafkaConsumerClient.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + await client.shutdown(); + + expect(mockConsumer.shutdown).toHaveBeenCalled(); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts new file mode 100644 index 0000000000..06a7b2594c --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts @@ -0,0 +1,67 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { Config } from '@backstage/config'; +import { EventsService } from '@backstage/plugin-events-node'; +import { Kafka } from 'kafkajs'; +import { KafkaEventSourceConfig, readConfig } from './config'; +import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; +import { LoggerServiceCreator } from './LoggerServiceCreator'; + +export class KafkaConsumerClient { + private readonly kafka: Kafka; + private readonly consumers: KafkaConsumingEventPublisher[]; + + static fromConfig(env: { + config: Config; + events: EventsService; + logger: LoggerService; + }): KafkaConsumerClient { + return new KafkaConsumerClient( + env.logger, + env.events, + readConfig(env.config), + ); + } + + private constructor( + logger: LoggerService, + events: EventsService, + config: KafkaEventSourceConfig, + ) { + this.kafka = new Kafka({ + ...config.kafkaConfig, + logCreator: LoggerServiceCreator(logger), + }); + + this.consumers = config.kafkaConsumerConfig.map(consumerConfig => + KafkaConsumingEventPublisher.fromConfig({ + kafkaClient: this.kafka, + config: consumerConfig, + logger, + events, + }), + ); + } + + async start(): Promise { + this.consumers.map(async consumer => await consumer.start()); + } + + async shutdown(): Promise { + this.consumers.map(async consumer => await consumer.shutdown()); + } +} diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts new file mode 100644 index 0000000000..c0bf94d3c8 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts @@ -0,0 +1,92 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; +import { Kafka } from 'kafkajs'; +import { KafkaConsumerConfig } from './config'; +import { mockServices } from '@backstage/backend-test-utils'; + +jest.mock('kafkajs'); + +describe('KafkaConsumingEventPublisher', () => { + const mockLogger = mockServices.logger.mock(); + const mockEvents = mockServices.events.mock(); + + const mockConsumer = { + connect: jest.fn(), + disconnect: jest.fn(), + subscribe: jest.fn(), + run: jest.fn(), + }; + + const mockKafkaClient = { + consumer: jest.fn().mockReturnValue(mockConsumer), + } as unknown as Kafka; + + const kafkaConsumerConfig: KafkaConsumerConfig = { + consumerConfig: { + groupId: 'test-group', + }, + consumerSubscribeConfig: { + topics: ['test-topic'], + }, + backstageTopic: 'backstage-topic', + }; + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('should create an instance from config', () => { + const consumer = KafkaConsumingEventPublisher.fromConfig({ + kafkaClient: mockKafkaClient, + config: kafkaConsumerConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(consumer).toBeInstanceOf(KafkaConsumingEventPublisher); + }); + + it('should start the consumer', async () => { + const consumer = KafkaConsumingEventPublisher.fromConfig({ + kafkaClient: mockKafkaClient, + config: kafkaConsumerConfig, + events: mockEvents, + logger: mockLogger, + }); + + await consumer.start(); + + expect(mockConsumer.connect).toHaveBeenCalled(); + expect(mockConsumer.subscribe).toHaveBeenCalledWith( + kafkaConsumerConfig.consumerSubscribeConfig, + ); + expect(mockConsumer.run).toHaveBeenCalled(); + }); + + it('should shutdown the consumer', async () => { + const consumer = KafkaConsumingEventPublisher.fromConfig({ + kafkaClient: mockKafkaClient, + config: kafkaConsumerConfig, + events: mockEvents, + logger: mockLogger, + }); + + await consumer.shutdown(); + + expect(mockConsumer.disconnect).toHaveBeenCalled(); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts new file mode 100644 index 0000000000..90db289a8a --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts @@ -0,0 +1,104 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { EventParams, EventsService } from '@backstage/plugin-events-node'; +import { Consumer, ConsumerSubscribeTopics, IHeaders, Kafka } from 'kafkajs'; +import { KafkaConsumerConfig } from './config'; + +type EventMetadata = EventParams['metadata']; + +export class KafkaConsumingEventPublisher { + private readonly kafkaConsumer: Consumer; + private readonly consumerSubscribeOptions: ConsumerSubscribeTopics; + private readonly backstageTopic: string; + private readonly logger: LoggerService; + + static fromConfig(env: { + kafkaClient: Kafka; + config: KafkaConsumerConfig; + events: EventsService; + logger: LoggerService; + }): KafkaConsumingEventPublisher { + return new KafkaConsumingEventPublisher( + env.kafkaClient, + env.logger, + env.events, + env.config, + ); + } + + private constructor( + kafkaClient: Kafka, + logger: LoggerService, + private readonly events: EventsService, + config: KafkaConsumerConfig, + ) { + this.kafkaConsumer = kafkaClient.consumer(config.consumerConfig); + this.consumerSubscribeOptions = config.consumerSubscribeConfig; + this.backstageTopic = config.backstageTopic; + const id = `events.kafka.publisher:${this.backstageTopic}`; + this.logger = logger.child({ + class: KafkaConsumingEventPublisher.prototype.constructor.name, + groupId: config.consumerConfig.groupId, + kafkaTopics: config.consumerSubscribeConfig.topics.toString(), + backstageTopic: config.backstageTopic, + taskId: id, + }); + } + + async start(): Promise { + try { + await this.kafkaConsumer.connect(); + this.logger.info('Kafka consumer connected'); + + await this.kafkaConsumer.subscribe(this.consumerSubscribeOptions); + + await this.kafkaConsumer.run({ + eachMessage: async ({ message }) => { + this.events.publish({ + topic: this.backstageTopic, + eventPayload: JSON.parse(message.value?.toString()!), + metadata: this.convertHeadersToMetadata(message.headers), + }); + }, + }); + } catch (error: any) { + this.logger.error('Kafka consumer connection failed ', error); + } + } + + async shutdown(): Promise { + await this.kafkaConsumer.disconnect(); + this.logger.info('Kafka consumer disconnected'); + } + + private convertHeadersToMetadata = ( + headers: IHeaders | undefined, + ): EventParams['metadata'] => { + if (!headers) return undefined; + + const metadata: EventMetadata = {}; + + Object.entries(headers).forEach(([key, value]) => { + // If value is an array use toString() on all values converting any Buffer types to valid strings + if (Array.isArray(value)) metadata[key] = value.map(v => v.toString()); + // Always return the values using toString() to catch all Buffer types that should be converted to strings + else metadata[key] = value?.toString(); + }); + + return metadata; + }; +} diff --git a/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts new file mode 100644 index 0000000000..0dd5f9cd84 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts @@ -0,0 +1,45 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { LogEntry, logLevel } from 'kafkajs'; + +export const LoggerServiceCreator = + (loggerService: LoggerService) => (_level: logLevel) => { + return (entry: LogEntry) => { + const { namespace, level, log } = entry; + const { message, ...extra } = log; + + const logMethods: Record< + logLevel, + (message: string, meta?: object) => void + > = { + [logLevel.ERROR]: loggerService.error, + [logLevel.WARN]: loggerService.warn, + [logLevel.INFO]: loggerService.info, + [logLevel.DEBUG]: loggerService.debug, + [logLevel.NOTHING]: () => {}, + }; + + // Use loggerService method that matches the level + logMethods[level].call( + loggerService, + `Kafka ${namespace} ${log.message}`, + { + ...extra, + }, + ); + }; + }; diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts new file mode 100644 index 0000000000..88c1d9747c --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts @@ -0,0 +1,197 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { ConfigReader } from '@backstage/config'; +import { readConfig } from './config'; + +describe('readConfig', () => { + it('not configured', () => { + const config = new ConfigReader({}); + + expect(() => { + readConfig(config); + }).toThrow(); + }); + + it('only required fields configured', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: [ + { + topic: 'fake1', + kafka: { + topics: ['topic-A'], + groupId: 'my-group', + }, + }, + { + topic: 'fake2', + kafka: { + topics: ['topic-B'], + groupId: 'my-group', + }, + }, + ], + }, + }, + }, + }, + }); + + const publisherConfigs = readConfig(config); + + expect(publisherConfigs.kafkaConsumerConfig.length).toBe(2); + + expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events'); + expect(publisherConfigs.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual( + 'fake1', + ); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId, + ).toEqual('my-group'); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.topics, + ).toEqual(['topic-A']); + }); + + it('all fields configured', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + ssl: true, + sasl: { + mechanism: 'plain', + username: 'username', + password: 'password', + }, + retry: { + maxRetryTime: '20000', + initialRetryTime: '200', + factor: '0.4', + multiplier: '4', + retries: '10', + }, + authenticationTimeout: 20000, + connectionTimeout: 1500, + requestTimeout: 20000, + enforceRequestTimeout: false, + topics: [ + { + topic: 'fake1', + kafka: { + topics: ['topic-A'], + groupId: 'my-group', + sessionTimeout: 20000, + rebalanceTimeout: 50000, + heartbeatInterval: 2000, + metadataMaxAge: 400000, + maxBytesPerPartition: 50000, + minBytes: 2, + maxBytes: 500000, + maxWaitTimeInMs: 4000, + }, + }, + { + topic: 'fake2', + kafka: { + topics: ['topic-B'], + groupId: 'my-group', + }, + }, + ], + }, + }, + }, + }, + }); + + const publisherConfigs = readConfig(config); + + // Client configuration + expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events'); + expect(publisherConfigs.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(publisherConfigs.kafkaConfig.ssl).toBeTruthy(); + expect(publisherConfigs.kafkaConfig.sasl).toStrictEqual({ + mechanism: 'plain', + username: 'username', + password: 'password', + }); + expect(publisherConfigs.kafkaConfig.authenticationTimeout).toBe(20000); + expect(publisherConfigs.kafkaConfig.connectionTimeout).toBe(1500); + expect(publisherConfigs.kafkaConfig.requestTimeout).toBe(20000); + expect(publisherConfigs.kafkaConfig.enforceRequestTimeout).toBeFalsy(); + expect(publisherConfigs.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: '20000', + initialRetryTime: '200', + factor: '0.4', + multiplier: '4', + retries: '10', + }); + + // Consumer configuration + expect(publisherConfigs.kafkaConsumerConfig.length).toBe(2); + expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual( + 'fake1', + ); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId, + ).toEqual('my-group'); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.topics, + ).toEqual(['topic-A']); + + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.sessionTimeout, + ).toBe(20000); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.rebalanceTimeout, + ).toBe(50000); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.heartbeatInterval, + ).toBe(2000); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.metadataMaxAge, + ).toBe(400000); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig + .maxBytesPerPartition, + ).toBe(50000); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.minBytes, + ).toBe(2); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxBytes, + ).toBe(500000); + expect( + publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxWaitTimeInMs, + ).toBe(4000); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts new file mode 100644 index 0000000000..206ea0b8ac --- /dev/null +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -0,0 +1,90 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs'; + +export interface KafkaConsumerConfig { + backstageTopic: string; + consumerConfig: ConsumerConfig; + consumerSubscribeConfig: ConsumerSubscribeTopics; +} + +export interface KafkaEventSourceConfig { + kafkaConfig: KafkaConfig; + kafkaConsumerConfig: KafkaConsumerConfig[]; +} + +const CONFIG_PREFIX_PUBLISHER = + 'events.modules.kafka.kafkaConsumingEventPublisher'; + +export const readConfig = (config: Config): KafkaEventSourceConfig => { + const kafkaConfig = config.getConfig(CONFIG_PREFIX_PUBLISHER); + + const clientId = kafkaConfig.getString('clientId'); + const brokers = kafkaConfig.getStringArray('brokers'); + + const authenticationTimeout = kafkaConfig.getOptionalNumber( + 'authenticationTimeout', + ); + const connectionTimeout = kafkaConfig.getOptionalNumber('connectionTimeout'); + const requestTimeout = kafkaConfig.getOptionalNumber('requestTimeout'); + const enforceRequestTimeout = kafkaConfig.getOptionalBoolean( + 'enforceRequestTimeout', + ); + + const ssl = kafkaConfig.getOptional('ssl') as KafkaConfig['ssl']; + const sasl = kafkaConfig.getOptional('sasl') as KafkaConfig['sasl']; + const retry = kafkaConfig.getOptional('retry') as KafkaConfig['retry']; + + const kafkaConsumerConfig: KafkaConsumerConfig[] = kafkaConfig + .getConfigArray('topics') + .map(topic => { + return { + backstageTopic: topic.getString('topic'), + consumerConfig: { + groupId: topic.getString('kafka.groupId'), + sessionTimeout: topic.getOptionalNumber('kafka.sessionTimeout'), + rebalanceTimeout: topic.getOptionalNumber('kafka.rebalanceTimeout'), + heartbeatInterval: topic.getOptionalNumber('kafka.heartbeatInterval'), + metadataMaxAge: topic.getOptionalNumber('kafka.metadataMaxAge'), + maxBytesPerPartition: topic.getOptionalNumber( + 'kafka.maxBytesPerPartition', + ), + minBytes: topic.getOptionalNumber('kafka.minBytes'), + maxBytes: topic.getOptionalNumber('kafka.maxBytes'), + maxWaitTimeInMs: topic.getOptionalNumber('kafka.maxWaitTimeInMs'), + }, + consumerSubscribeConfig: { + topics: topic.getStringArray('kafka.topics'), + }, + }; + }); + + return { + kafkaConfig: { + clientId, + brokers, + ssl, + sasl, + authenticationTimeout, + connectionTimeout, + requestTimeout, + enforceRequestTimeout, + retry, + }, + kafkaConsumerConfig, + }; +}; diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts new file mode 100644 index 0000000000..816793f635 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts @@ -0,0 +1,85 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { createServiceFactory } from '@backstage/backend-plugin-api'; +import { mockServices, startTestBackend } from '@backstage/backend-test-utils'; +import { eventsServiceRef } from '@backstage/plugin-events-node'; +import { TestEventsService } from '@backstage/plugin-events-backend-test-utils'; +import { eventsModuleKafkaConsumingEventPublisher } from './eventsModuleKafkaConsumingEventPublisher'; +import { KafkaConsumerClient } from '../publisher/KafkaConsumerClient'; + +jest.mock('../publisher/KafkaConsumerClient'); + +describe('eventsModuleKafkaConsumingEventPublisher', () => { + it('should be correctly wired and set up', async () => { + const events = new TestEventsService(); + const eventsServiceFactory = createServiceFactory({ + service: eventsServiceRef, + deps: {}, + async factory({}) { + return events; + }, + }); + + const mockKafkaConsumerClient = { + start: jest.fn(), + shutdown: jest.fn(), + }; + (KafkaConsumerClient.fromConfig as jest.Mock).mockReturnValue( + mockKafkaConsumerClient, + ); + + await startTestBackend({ + features: [ + eventsServiceFactory, + eventsModuleKafkaConsumingEventPublisher, + mockServices.rootConfig.factory({ + data: { + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: { + fake1: { + kafka: { + topics: ['topic-A'], + groupId: 'my-group', + }, + }, + fake2: { + kafka: { + topics: ['topic-B'], + groupId: 'my-group', + }, + }, + }, + }, + }, + }, + }, + }, + }), + ], + }); + + // Verify that the Kafka consumer client was started + expect(mockKafkaConsumerClient.start).toHaveBeenCalled(); + + // Verify that the shutdown hook was registered + expect(mockKafkaConsumerClient.shutdown).not.toHaveBeenCalled(); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts new file mode 100644 index 0000000000..a458a39662 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts @@ -0,0 +1,52 @@ +/* + * Copyright 2025 The Backstage Authors + * + * 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 { + coreServices, + createBackendModule, +} from '@backstage/backend-plugin-api'; +import { KafkaConsumerClient } from '../publisher/KafkaConsumerClient'; +import { eventsServiceRef } from '@backstage/plugin-events-node'; + +/** + * Kafka module for the Events plugin. + * + * @public + */ +export const eventsModuleKafkaConsumingEventPublisher = createBackendModule({ + pluginId: 'events', + moduleId: 'kafka-consuming-event-publisher', + register(env) { + env.registerInit({ + deps: { + config: coreServices.rootConfig, + events: eventsServiceRef, + logger: coreServices.logger, + lifecycle: coreServices.lifecycle, + }, + async init({ config, logger, events, lifecycle }) { + const kafka = KafkaConsumerClient.fromConfig({ + config, + events, + logger, + }); + + await kafka.start(); + + lifecycle.addShutdownHook(async () => await kafka.shutdown()); + }, + }); + }, +}); diff --git a/yarn.lock b/yarn.lock index 1e822bfc7c..b1dea974cb 100644 --- a/yarn.lock +++ b/yarn.lock @@ -35254,6 +35254,13 @@ __metadata: languageName: node linkType: hard +"kafkajs@npm:^2.2.4": + version: 2.2.4 + resolution: "kafkajs@npm:2.2.4" + checksum: 83e9e8bc50a09b142f4ff79f6a2bd88ecc21b83bcefe6621ab1716118d624886befb7371731274f67812ce35dd50b53140ff3b49a06e5d9169fe6b164d72fea5 + languageName: node + linkType: hard + "keygrip@npm:~1.1.0": version: 1.1.0 resolution: "keygrip@npm:1.1.0" From 89658c5ae463fb16928a18baf01970a2a5e67237 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 19 Mar 2025 14:34:40 +0100 Subject: [PATCH 02/15] chore(api-report): remove api report warnings Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/report.api.md | 25 +++++++++++++++---- .../events-backend-module-kafka/src/index.ts | 13 ++++++++++ .../src/publisher/KafkaConsumerClient.ts | 7 ++++++ .../publisher/KafkaConsumingEventPublisher.ts | 7 ++++++ .../src/publisher/config.ts | 6 +++++ 5 files changed, 53 insertions(+), 5 deletions(-) diff --git a/plugins/events-backend-module-kafka/report.api.md b/plugins/events-backend-module-kafka/report.api.md index b8fd142d0e..6c2c4c70f4 100644 --- a/plugins/events-backend-module-kafka/report.api.md +++ b/plugins/events-backend-module-kafka/report.api.md @@ -1,4 +1,4 @@ -## API Report File for "@internal/plugin-events-backend-module-kafka" +## API Report File for "@backstage/plugin-events-backend-module-kafka" > Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). @@ -9,13 +9,14 @@ import { ConsumerConfig } from 'kafkajs'; import { ConsumerSubscribeTopics } from 'kafkajs'; import { EventsService } from '@backstage/plugin-events-node'; import { Kafka } from 'kafkajs'; +import { KafkaConfig } from 'kafkajs'; import { LoggerService } from '@backstage/backend-plugin-api'; // @public const eventsModuleKafkaConsumingEventPublisher: BackendFeature; export default eventsModuleKafkaConsumingEventPublisher; -// @public (undocumented) +// @public export class KafkaConsumerClient { // (undocumented) static fromConfig(env: { @@ -30,6 +31,16 @@ export class KafkaConsumerClient { } // @public (undocumented) +export interface KafkaConsumerConfig { + // (undocumented) + backstageTopic: string; + // (undocumented) + consumerConfig: ConsumerConfig; + // (undocumented) + consumerSubscribeConfig: ConsumerSubscribeTopics; +} + +// @public export class KafkaConsumingEventPublisher { // (undocumented) static fromConfig(env: { @@ -44,7 +55,11 @@ export class KafkaConsumingEventPublisher { start(): Promise; } -// Warnings were encountered during analysis: -// -// src/publisher/KafkaConsumingEventPublisher.d.ts:13:9 - (ae-forgotten-export) The symbol "KafkaConsumerConfig" needs to be exported by the entry point index.d.ts +// @public (undocumented) +export interface KafkaEventSourceConfig { + // (undocumented) + kafkaConfig: KafkaConfig; + // (undocumented) + kafkaConsumerConfig: KafkaConsumerConfig[]; +} ``` diff --git a/plugins/events-backend-module-kafka/src/index.ts b/plugins/events-backend-module-kafka/src/index.ts index ff4fe2a209..8980bcbaa7 100644 --- a/plugins/events-backend-module-kafka/src/index.ts +++ b/plugins/events-backend-module-kafka/src/index.ts @@ -14,6 +14,19 @@ * limitations under the License. */ +/** + * The module "kafka" for the Backstage backend plugin "events" + * adding an Kafka-based publisher, + * receiving events from an Kafka topic and passing it to the + * internal event broker. + * + * @packageDocumentation + */ + export { KafkaConsumerClient } from './publisher/KafkaConsumerClient'; export { KafkaConsumingEventPublisher } from './publisher/KafkaConsumingEventPublisher'; export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher'; +export type { + KafkaConsumerConfig, + KafkaEventSourceConfig, +} from './publisher/config'; diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts index 06a7b2594c..124eb4f1cb 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts @@ -21,6 +21,13 @@ import { KafkaEventSourceConfig, readConfig } from './config'; import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; import { LoggerServiceCreator } from './LoggerServiceCreator'; +/** + * KafkaConsumerClient + * + * This class creates the Kafka client that will be used to create the KafkaConsumingEventPublisher + * + * @public + */ export class KafkaConsumerClient { private readonly kafka: Kafka; private readonly consumers: KafkaConsumingEventPublisher[]; diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts index 90db289a8a..53df325af4 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts @@ -20,6 +20,13 @@ import { KafkaConsumerConfig } from './config'; type EventMetadata = EventParams['metadata']; +/** + * + * This class subscribes to Kafka topics and publishes events received to the registered subscriber. + * The message payload will be used as the event payload and passed to the subscribers. + * + * @public + */ export class KafkaConsumingEventPublisher { private readonly kafkaConsumer: Consumer; private readonly consumerSubscribeOptions: ConsumerSubscribeTopics; diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts index 206ea0b8ac..b62598ab80 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -16,12 +16,18 @@ import { Config } from '@backstage/config'; import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs'; +/** + * @public + */ export interface KafkaConsumerConfig { backstageTopic: string; consumerConfig: ConsumerConfig; consumerSubscribeConfig: ConsumerSubscribeTopics; } +/** + * @public + */ export interface KafkaEventSourceConfig { kafkaConfig: KafkaConfig; kafkaConsumerConfig: KafkaConsumerConfig[]; From b034b9d7d5ce6724cf32a21e48aad9e17706e6a0 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 19 Mar 2025 14:55:26 +0100 Subject: [PATCH 03/15] chore(changeset): add changeset Signed-off-by: Jonas Beck --- .changeset/deep-aliens-camp.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .changeset/deep-aliens-camp.md diff --git a/.changeset/deep-aliens-camp.md b/.changeset/deep-aliens-camp.md new file mode 100644 index 0000000000..ca5c3d8ed4 --- /dev/null +++ b/.changeset/deep-aliens-camp.md @@ -0,0 +1,7 @@ +--- +'@backstage/plugin-events-backend-module-kafka': minor +--- + +Adds a new module `kafka` for plugin-events-backend + +The module introduces the `KafkaConsumerClient` which creates a Kafka client used to establish consumer connections. It also provides the `KafkaConsumingEventPublisher`, a consumer that subscribes to configured Kafka topics and publishes received messages to the Event Service. From 48e632280b84cd99a2d7f0955783c1b443e78673 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Thu, 20 Mar 2025 09:36:07 +0100 Subject: [PATCH 04/15] chore: update package.json to follow other examples Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/package.json | 37 ++++++++++++------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/plugins/events-backend-module-kafka/package.json b/plugins/events-backend-module-kafka/package.json index 1750caf36b..4b9594c4af 100644 --- a/plugins/events-backend-module-kafka/package.json +++ b/plugins/events-backend-module-kafka/package.json @@ -1,27 +1,38 @@ { "name": "@backstage/plugin-events-backend-module-kafka", - "description": "The kafka backend module for the events plugin.", "version": "0.0.0", - "main": "src/index.ts", - "types": "src/index.ts", - "license": "Apache-2.0", - "private": true, + "description": "The kafka backend module for the events plugin.", + "backstage": { + "role": "backend-plugin-module", + "pluginId": "events", + "pluginPackage": "@backstage/plugin-events-backend" + }, "publishConfig": { "access": "public", "main": "dist/index.cjs.js", "types": "dist/index.d.ts" }, - "backstage": { - "role": "backend-plugin-module" + "homepage": "https://backstage.io", + "repository": { + "type": "git", + "url": "https://github.com/backstage/backstage", + "directory": "plugins/events-backend-module-kafka" }, + "license": "Apache-2.0", + "main": "src/index.ts", + "types": "src/index.ts", + "files": [ + "dist", + "config.d.ts" + ], "scripts": { - "start": "backstage-cli package start", "build": "backstage-cli package build", - "lint": "backstage-cli package lint", - "test": "backstage-cli package test", "clean": "backstage-cli package clean", + "lint": "backstage-cli package lint", "prepack": "backstage-cli package prepack", - "postpack": "backstage-cli package postpack" + "postpack": "backstage-cli package postpack", + "start": "backstage-cli package start", + "test": "backstage-cli package test" }, "dependencies": { "@backstage/backend-plugin-api": "workspace:^", @@ -35,7 +46,5 @@ "@backstage/cli": "workspace:^", "@backstage/plugin-events-backend-test-utils": "workspace:^" }, - "files": [ - "dist" - ] + "configSchema": "config.d.ts" } From 4da644908a2db337981fd1dfadc59525eb455088 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Thu, 20 Mar 2025 09:53:25 +0100 Subject: [PATCH 05/15] docs(readme): update config.d.ts path Signed-off-by: Jonas Beck --- plugins/events-backend-module-kafka/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/events-backend-module-kafka/README.md b/plugins/events-backend-module-kafka/README.md index 0bdbd5cf1e..dbc5bf5607 100644 --- a/plugins/events-backend-module-kafka/README.md +++ b/plugins/events-backend-module-kafka/README.md @@ -27,7 +27,7 @@ events: groupId: your-group-id # (Required) The GroupId to be used by the topic consumers. ``` -For a complete list of all available fields that can be configured, refer to the [config.d.ts file](path/to/config.d.ts). +For a complete list of all available fields that can be configured, refer to the [config.d.ts file](./config.d.ts). ### Optional SSL Configuration From 70828101f96dd9272d5ae667053884adff9a5a70 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Tue, 1 Apr 2025 09:52:18 +0200 Subject: [PATCH 06/15] chore(events): update kafkaConsumerConfig name Signed-off-by: Jonas Beck --- .../src/publisher/KafkaConsumerClient.ts | 2 +- .../src/publisher/config.test.ts | 32 +++++++++---------- .../src/publisher/config.ts | 6 ++-- 3 files changed, 20 insertions(+), 20 deletions(-) diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts index 124eb4f1cb..16c2734aef 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts @@ -54,7 +54,7 @@ export class KafkaConsumerClient { logCreator: LoggerServiceCreator(logger), }); - this.consumers = config.kafkaConsumerConfig.map(consumerConfig => + this.consumers = config.kafkaConsumerConfigs.map(consumerConfig => KafkaConsumingEventPublisher.fromConfig({ kafkaClient: this.kafka, config: consumerConfig, diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts index 88c1d9747c..083c206c77 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts @@ -57,21 +57,21 @@ describe('readConfig', () => { const publisherConfigs = readConfig(config); - expect(publisherConfigs.kafkaConsumerConfig.length).toBe(2); + expect(publisherConfigs.kafkaConsumerConfigs.length).toBe(2); expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events'); expect(publisherConfigs.kafkaConfig.brokers).toEqual([ 'kafka1:9092', 'kafka2:9092', ]); - expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual( + expect(publisherConfigs.kafkaConsumerConfigs[0].backstageTopic).toEqual( 'fake1', ); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.groupId, ).toEqual('my-group'); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.topics, + publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeConfig.topics, ).toEqual(['topic-A']); }); @@ -157,41 +157,41 @@ describe('readConfig', () => { }); // Consumer configuration - expect(publisherConfigs.kafkaConsumerConfig.length).toBe(2); - expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual( + expect(publisherConfigs.kafkaConsumerConfigs.length).toBe(2); + expect(publisherConfigs.kafkaConsumerConfigs[0].backstageTopic).toEqual( 'fake1', ); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.groupId, ).toEqual('my-group'); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.topics, + publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeConfig.topics, ).toEqual(['topic-A']); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.sessionTimeout, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.sessionTimeout, ).toBe(20000); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.rebalanceTimeout, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.rebalanceTimeout, ).toBe(50000); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.heartbeatInterval, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.heartbeatInterval, ).toBe(2000); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.metadataMaxAge, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.metadataMaxAge, ).toBe(400000); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig .maxBytesPerPartition, ).toBe(50000); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.minBytes, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.minBytes, ).toBe(2); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxBytes, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.maxBytes, ).toBe(500000); expect( - publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxWaitTimeInMs, + publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.maxWaitTimeInMs, ).toBe(4000); }); }); diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts index b62598ab80..f5e4379edd 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -30,7 +30,7 @@ export interface KafkaConsumerConfig { */ export interface KafkaEventSourceConfig { kafkaConfig: KafkaConfig; - kafkaConsumerConfig: KafkaConsumerConfig[]; + kafkaConsumerConfigs: KafkaConsumerConfig[]; } const CONFIG_PREFIX_PUBLISHER = @@ -55,7 +55,7 @@ export const readConfig = (config: Config): KafkaEventSourceConfig => { const sasl = kafkaConfig.getOptional('sasl') as KafkaConfig['sasl']; const retry = kafkaConfig.getOptional('retry') as KafkaConfig['retry']; - const kafkaConsumerConfig: KafkaConsumerConfig[] = kafkaConfig + const kafkaConsumerConfigs: KafkaConsumerConfig[] = kafkaConfig .getConfigArray('topics') .map(topic => { return { @@ -91,6 +91,6 @@ export const readConfig = (config: Config): KafkaEventSourceConfig => { enforceRequestTimeout, retry, }, - kafkaConsumerConfig, + kafkaConsumerConfigs, }; }; From 500dbad885ee0582b10600a5ce473abe2bb138e2 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Tue, 1 Apr 2025 10:31:37 +0200 Subject: [PATCH 07/15] chore(events): update consumerSubscribeOptions name Signed-off-by: Jonas Beck --- .../src/publisher/KafkaConsumingEventPublisher.test.ts | 4 ++-- .../src/publisher/KafkaConsumingEventPublisher.ts | 8 ++++---- .../src/publisher/config.test.ts | 4 ++-- .../events-backend-module-kafka/src/publisher/config.ts | 4 ++-- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts index c0bf94d3c8..64c53595ad 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts @@ -39,7 +39,7 @@ describe('KafkaConsumingEventPublisher', () => { consumerConfig: { groupId: 'test-group', }, - consumerSubscribeConfig: { + consumerSubscribeTopics: { topics: ['test-topic'], }, backstageTopic: 'backstage-topic', @@ -72,7 +72,7 @@ describe('KafkaConsumingEventPublisher', () => { expect(mockConsumer.connect).toHaveBeenCalled(); expect(mockConsumer.subscribe).toHaveBeenCalledWith( - kafkaConsumerConfig.consumerSubscribeConfig, + kafkaConsumerConfig.consumerSubscribeTopics, ); expect(mockConsumer.run).toHaveBeenCalled(); }); diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts index 53df325af4..cc794481cf 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts @@ -29,7 +29,7 @@ type EventMetadata = EventParams['metadata']; */ export class KafkaConsumingEventPublisher { private readonly kafkaConsumer: Consumer; - private readonly consumerSubscribeOptions: ConsumerSubscribeTopics; + private readonly consumerSubscribeTopics: ConsumerSubscribeTopics; private readonly backstageTopic: string; private readonly logger: LoggerService; @@ -54,13 +54,13 @@ export class KafkaConsumingEventPublisher { config: KafkaConsumerConfig, ) { this.kafkaConsumer = kafkaClient.consumer(config.consumerConfig); - this.consumerSubscribeOptions = config.consumerSubscribeConfig; + this.consumerSubscribeTopics = config.consumerSubscribeTopics; this.backstageTopic = config.backstageTopic; const id = `events.kafka.publisher:${this.backstageTopic}`; this.logger = logger.child({ class: KafkaConsumingEventPublisher.prototype.constructor.name, groupId: config.consumerConfig.groupId, - kafkaTopics: config.consumerSubscribeConfig.topics.toString(), + kafkaTopics: config.consumerSubscribeTopics.topics.toString(), backstageTopic: config.backstageTopic, taskId: id, }); @@ -71,7 +71,7 @@ export class KafkaConsumingEventPublisher { await this.kafkaConsumer.connect(); this.logger.info('Kafka consumer connected'); - await this.kafkaConsumer.subscribe(this.consumerSubscribeOptions); + await this.kafkaConsumer.subscribe(this.consumerSubscribeTopics); await this.kafkaConsumer.run({ eachMessage: async ({ message }) => { diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts index 083c206c77..6381880437 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts @@ -71,7 +71,7 @@ describe('readConfig', () => { publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.groupId, ).toEqual('my-group'); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeConfig.topics, + publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, ).toEqual(['topic-A']); }); @@ -165,7 +165,7 @@ describe('readConfig', () => { publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.groupId, ).toEqual('my-group'); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeConfig.topics, + publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, ).toEqual(['topic-A']); expect( diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts index f5e4379edd..b872a2a8d9 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -22,7 +22,7 @@ import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs'; export interface KafkaConsumerConfig { backstageTopic: string; consumerConfig: ConsumerConfig; - consumerSubscribeConfig: ConsumerSubscribeTopics; + consumerSubscribeTopics: ConsumerSubscribeTopics; } /** @@ -73,7 +73,7 @@ export const readConfig = (config: Config): KafkaEventSourceConfig => { maxBytes: topic.getOptionalNumber('kafka.maxBytes'), maxWaitTimeInMs: topic.getOptionalNumber('kafka.maxWaitTimeInMs'), }, - consumerSubscribeConfig: { + consumerSubscribeTopics: { topics: topic.getStringArray('kafka.topics'), }, }; From 213e295f9a53394ca05a70f08385801edffd4a9c Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Tue, 1 Apr 2025 10:33:25 +0200 Subject: [PATCH 08/15] chore(events): rename loggerServiceCreator Signed-off-by: Jonas Beck --- .../src/publisher/KafkaConsumerClient.ts | 4 ++-- .../{LoggerServiceCreator.ts => LoggerServiceAdapter.ts} | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) rename plugins/events-backend-module-kafka/src/publisher/{LoggerServiceCreator.ts => LoggerServiceAdapter.ts} (97%) diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts index 16c2734aef..5b6366b84d 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts @@ -19,7 +19,7 @@ import { EventsService } from '@backstage/plugin-events-node'; import { Kafka } from 'kafkajs'; import { KafkaEventSourceConfig, readConfig } from './config'; import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; -import { LoggerServiceCreator } from './LoggerServiceCreator'; +import { LoggerServiceAdapter } from './LoggerServiceAdapter'; /** * KafkaConsumerClient @@ -51,7 +51,7 @@ export class KafkaConsumerClient { ) { this.kafka = new Kafka({ ...config.kafkaConfig, - logCreator: LoggerServiceCreator(logger), + logCreator: LoggerServiceAdapter(logger), }); this.consumers = config.kafkaConsumerConfigs.map(consumerConfig => diff --git a/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts similarity index 97% rename from plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts rename to plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts index 0dd5f9cd84..3750d0c8a1 100644 --- a/plugins/events-backend-module-kafka/src/publisher/LoggerServiceCreator.ts +++ b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts @@ -16,7 +16,7 @@ import { LoggerService } from '@backstage/backend-plugin-api'; import { LogEntry, logLevel } from 'kafkajs'; -export const LoggerServiceCreator = +export const LoggerServiceAdapter = (loggerService: LoggerService) => (_level: logLevel) => { return (entry: LogEntry) => { const { namespace, level, log } = entry; From 77db4f11f8e123e4ff06291bbbbf1daeb578df75 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Tue, 1 Apr 2025 10:35:08 +0200 Subject: [PATCH 09/15] chore(events): reduce api surface Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/report.api.md | 54 ------------------- .../events-backend-module-kafka/src/index.ts | 6 --- 2 files changed, 60 deletions(-) diff --git a/plugins/events-backend-module-kafka/report.api.md b/plugins/events-backend-module-kafka/report.api.md index 6c2c4c70f4..a6b129b82e 100644 --- a/plugins/events-backend-module-kafka/report.api.md +++ b/plugins/events-backend-module-kafka/report.api.md @@ -4,62 +4,8 @@ ```ts import { BackendFeature } from '@backstage/backend-plugin-api'; -import { Config } from '@backstage/config'; -import { ConsumerConfig } from 'kafkajs'; -import { ConsumerSubscribeTopics } from 'kafkajs'; -import { EventsService } from '@backstage/plugin-events-node'; -import { Kafka } from 'kafkajs'; -import { KafkaConfig } from 'kafkajs'; -import { LoggerService } from '@backstage/backend-plugin-api'; // @public const eventsModuleKafkaConsumingEventPublisher: BackendFeature; export default eventsModuleKafkaConsumingEventPublisher; - -// @public -export class KafkaConsumerClient { - // (undocumented) - static fromConfig(env: { - config: Config; - events: EventsService; - logger: LoggerService; - }): KafkaConsumerClient; - // (undocumented) - shutdown(): Promise; - // (undocumented) - start(): Promise; -} - -// @public (undocumented) -export interface KafkaConsumerConfig { - // (undocumented) - backstageTopic: string; - // (undocumented) - consumerConfig: ConsumerConfig; - // (undocumented) - consumerSubscribeConfig: ConsumerSubscribeTopics; -} - -// @public -export class KafkaConsumingEventPublisher { - // (undocumented) - static fromConfig(env: { - kafkaClient: Kafka; - config: KafkaConsumerConfig; - events: EventsService; - logger: LoggerService; - }): KafkaConsumingEventPublisher; - // (undocumented) - shutdown(): Promise; - // (undocumented) - start(): Promise; -} - -// @public (undocumented) -export interface KafkaEventSourceConfig { - // (undocumented) - kafkaConfig: KafkaConfig; - // (undocumented) - kafkaConsumerConfig: KafkaConsumerConfig[]; -} ``` diff --git a/plugins/events-backend-module-kafka/src/index.ts b/plugins/events-backend-module-kafka/src/index.ts index 8980bcbaa7..3ba6207d7d 100644 --- a/plugins/events-backend-module-kafka/src/index.ts +++ b/plugins/events-backend-module-kafka/src/index.ts @@ -23,10 +23,4 @@ * @packageDocumentation */ -export { KafkaConsumerClient } from './publisher/KafkaConsumerClient'; -export { KafkaConsumingEventPublisher } from './publisher/KafkaConsumingEventPublisher'; export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher'; -export type { - KafkaConsumerConfig, - KafkaEventSourceConfig, -} from './publisher/config'; From d57ef9e0350e3135d2fe41e841f0432e26d94a31 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 23 Apr 2025 16:05:37 +0200 Subject: [PATCH 10/15] chore(yarn): update yarn.lock file Signed-off-by: Jonas Beck --- yarn.lock | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/yarn.lock b/yarn.lock index b1dea974cb..11e3bd9ae4 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6635,6 +6635,21 @@ __metadata: languageName: unknown linkType: soft +"@backstage/plugin-events-backend-module-kafka@workspace:plugins/events-backend-module-kafka": + version: 0.0.0-use.local + resolution: "@backstage/plugin-events-backend-module-kafka@workspace:plugins/events-backend-module-kafka" + dependencies: + "@backstage/backend-plugin-api": "workspace:^" + "@backstage/backend-test-utils": "workspace:^" + "@backstage/cli": "workspace:^" + "@backstage/config": "workspace:^" + "@backstage/plugin-events-backend-test-utils": "workspace:^" + "@backstage/plugin-events-node": "workspace:^" + "@backstage/types": "workspace:^" + kafkajs: "npm:^2.2.4" + languageName: unknown + linkType: soft + "@backstage/plugin-events-backend-test-utils@workspace:^, @backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils": version: 0.0.0-use.local resolution: "@backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils" @@ -35257,7 +35272,7 @@ __metadata: "kafkajs@npm:^2.2.4": version: 2.2.4 resolution: "kafkajs@npm:2.2.4" - checksum: 83e9e8bc50a09b142f4ff79f6a2bd88ecc21b83bcefe6621ab1716118d624886befb7371731274f67812ce35dd50b53140ff3b49a06e5d9169fe6b164d72fea5 + checksum: 10/75eb0d221397085f90e51f8a2d752495c9fa9c1b3a1a6db610cd7074fa8c52777f295832fd0a7c49cded5e574337a09fafa8c3f7cf1caa38f4dc9aa20fcfb7df languageName: node linkType: hard From 946f88f0ca6402a6ea4c76351616270727bb2975 Mon Sep 17 00:00:00 2001 From: Jonas Beck <112394761+Jonas-Beck@users.noreply.github.com> Date: Wed, 4 Jun 2025 08:51:55 +0200 Subject: [PATCH 11/15] Update plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Signed-off-by: Jonas Beck <112394761+Jonas-Beck@users.noreply.github.com> --- .../src/publisher/KafkaConsumerClient.ts | 4 ++-- .../src/publisher/LoggerServiceAdapter.ts | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts index 5b6366b84d..762a95dd41 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts @@ -19,7 +19,7 @@ import { EventsService } from '@backstage/plugin-events-node'; import { Kafka } from 'kafkajs'; import { KafkaEventSourceConfig, readConfig } from './config'; import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; -import { LoggerServiceAdapter } from './LoggerServiceAdapter'; +import { loggerServiceAdapter } from './LoggerServiceAdapter'; /** * KafkaConsumerClient @@ -51,7 +51,7 @@ export class KafkaConsumerClient { ) { this.kafka = new Kafka({ ...config.kafkaConfig, - logCreator: LoggerServiceAdapter(logger), + logCreator: loggerServiceAdapter(logger), }); this.consumers = config.kafkaConsumerConfigs.map(consumerConfig => diff --git a/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts index 3750d0c8a1..b9e9379104 100644 --- a/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts +++ b/plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts @@ -16,7 +16,7 @@ import { LoggerService } from '@backstage/backend-plugin-api'; import { LogEntry, logLevel } from 'kafkajs'; -export const LoggerServiceAdapter = +export const loggerServiceAdapter = (loggerService: LoggerService) => (_level: logLevel) => { return (entry: LogEntry) => { const { namespace, level, log } = entry; From 551daf9ebad2fc904cbfefe48c8722553980ae24 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 4 Jun 2025 08:57:34 +0200 Subject: [PATCH 12/15] chore(events): remove kafka consumer start and shutdown log Signed-off-by: Jonas Beck --- .../src/publisher/KafkaConsumingEventPublisher.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts index cc794481cf..223e4e99ac 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts @@ -69,7 +69,6 @@ export class KafkaConsumingEventPublisher { async start(): Promise { try { await this.kafkaConsumer.connect(); - this.logger.info('Kafka consumer connected'); await this.kafkaConsumer.subscribe(this.consumerSubscribeTopics); @@ -89,7 +88,6 @@ export class KafkaConsumingEventPublisher { async shutdown(): Promise { await this.kafkaConsumer.disconnect(); - this.logger.info('Kafka consumer disconnected'); } private convertHeadersToMetadata = ( From 9093d77867e6c28c7325d149210549687d8653c2 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 4 Jun 2025 09:37:25 +0200 Subject: [PATCH 13/15] chore(events): make config optional Signed-off-by: Jonas Beck --- .../src/publisher/KafkaConsumerClient.test.ts | 18 ++++- .../src/publisher/KafkaConsumerClient.ts | 19 ++++-- .../src/publisher/config.test.ts | 65 ++++++++++--------- .../src/publisher/config.ts | 10 ++- ...ventsModuleKafkaConsumingEventPublisher.ts | 4 ++ 5 files changed, 74 insertions(+), 42 deletions(-) diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts index c0d6809514..5824bf4d41 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts @@ -68,6 +68,16 @@ describe('KafkaConsumerClient', () => { expect(client).toBeInstanceOf(KafkaConsumerClient); }); + it('should not create an instance from config', () => { + const client = KafkaConsumerClient.fromConfig({ + config: new ConfigReader({}), + events: mockEvents, + logger: mockLogger, + }); + + expect(client).toBeUndefined(); + }); + it('should create a consumer for each topic from config', () => { KafkaConsumerClient.fromConfig({ config: mockConfig, @@ -92,7 +102,9 @@ describe('KafkaConsumerClient', () => { logger: mockLogger, }); - await client.start(); + expect(client).toBeDefined(); + + await client?.start(); expect(mockConsumer.start).toHaveBeenCalled(); }); @@ -111,7 +123,9 @@ describe('KafkaConsumerClient', () => { logger: mockLogger, }); - await client.shutdown(); + expect(client).toBeDefined(); + + await client?.shutdown(); expect(mockConsumer.shutdown).toHaveBeenCalled(); }); diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts index 762a95dd41..735b0fc224 100644 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts +++ b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts @@ -32,16 +32,21 @@ export class KafkaConsumerClient { private readonly kafka: Kafka; private readonly consumers: KafkaConsumingEventPublisher[]; - static fromConfig(env: { + static fromConfig(options: { config: Config; events: EventsService; logger: LoggerService; - }): KafkaConsumerClient { - return new KafkaConsumerClient( - env.logger, - env.events, - readConfig(env.config), - ); + }): KafkaConsumerClient | undefined { + const kafkaConfig = readConfig(options.config); + + if (!kafkaConfig) { + options.logger.info( + 'Kafka consumer not configured, skipping initialization', + ); + return undefined; + } + + return new KafkaConsumerClient(options.logger, options.events, kafkaConfig); } private constructor( diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts index 6381880437..5b2dfb6f0c 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts @@ -18,11 +18,9 @@ import { readConfig } from './config'; describe('readConfig', () => { it('not configured', () => { - const config = new ConfigReader({}); + const publisherConfigs = readConfig(new ConfigReader({})); - expect(() => { - readConfig(config); - }).toThrow(); + expect(publisherConfigs).toBeUndefined(); }); it('only required fields configured', () => { @@ -57,21 +55,23 @@ describe('readConfig', () => { const publisherConfigs = readConfig(config); - expect(publisherConfigs.kafkaConsumerConfigs.length).toBe(2); + expect(publisherConfigs).toBeDefined(); - expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events'); - expect(publisherConfigs.kafkaConfig.brokers).toEqual([ + expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(2); + + expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events'); + expect(publisherConfigs?.kafkaConfig.brokers).toEqual([ 'kafka1:9092', 'kafka2:9092', ]); - expect(publisherConfigs.kafkaConsumerConfigs[0].backstageTopic).toEqual( + expect(publisherConfigs?.kafkaConsumerConfigs[0].backstageTopic).toEqual( 'fake1', ); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.groupId, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.groupId, ).toEqual('my-group'); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, + publisherConfigs?.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, ).toEqual(['topic-A']); }); @@ -132,23 +132,25 @@ describe('readConfig', () => { const publisherConfigs = readConfig(config); + expect(publisherConfigs).toBeDefined(); + // Client configuration - expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events'); - expect(publisherConfigs.kafkaConfig.brokers).toEqual([ + expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events'); + expect(publisherConfigs?.kafkaConfig.brokers).toEqual([ 'kafka1:9092', 'kafka2:9092', ]); - expect(publisherConfigs.kafkaConfig.ssl).toBeTruthy(); - expect(publisherConfigs.kafkaConfig.sasl).toStrictEqual({ + expect(publisherConfigs?.kafkaConfig.ssl).toBeTruthy(); + expect(publisherConfigs?.kafkaConfig.sasl).toStrictEqual({ mechanism: 'plain', username: 'username', password: 'password', }); - expect(publisherConfigs.kafkaConfig.authenticationTimeout).toBe(20000); - expect(publisherConfigs.kafkaConfig.connectionTimeout).toBe(1500); - expect(publisherConfigs.kafkaConfig.requestTimeout).toBe(20000); - expect(publisherConfigs.kafkaConfig.enforceRequestTimeout).toBeFalsy(); - expect(publisherConfigs.kafkaConfig.retry).toStrictEqual({ + expect(publisherConfigs?.kafkaConfig.authenticationTimeout).toBe(20000); + expect(publisherConfigs?.kafkaConfig.connectionTimeout).toBe(1500); + expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(20000); + expect(publisherConfigs?.kafkaConfig.enforceRequestTimeout).toBeFalsy(); + expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({ maxRetryTime: '20000', initialRetryTime: '200', factor: '0.4', @@ -157,41 +159,42 @@ describe('readConfig', () => { }); // Consumer configuration - expect(publisherConfigs.kafkaConsumerConfigs.length).toBe(2); - expect(publisherConfigs.kafkaConsumerConfigs[0].backstageTopic).toEqual( + expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(2); + expect(publisherConfigs?.kafkaConsumerConfigs[0].backstageTopic).toEqual( 'fake1', ); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.groupId, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.groupId, ).toEqual('my-group'); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, + publisherConfigs?.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, ).toEqual(['topic-A']); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.sessionTimeout, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.sessionTimeout, ).toBe(20000); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.rebalanceTimeout, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.rebalanceTimeout, ).toBe(50000); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.heartbeatInterval, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig + .heartbeatInterval, ).toBe(2000); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.metadataMaxAge, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.metadataMaxAge, ).toBe(400000); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig .maxBytesPerPartition, ).toBe(50000); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.minBytes, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.minBytes, ).toBe(2); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.maxBytes, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxBytes, ).toBe(500000); expect( - publisherConfigs.kafkaConsumerConfigs[0].consumerConfig.maxWaitTimeInMs, + publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxWaitTimeInMs, ).toBe(4000); }); }); diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts index b872a2a8d9..aadf4a5166 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -36,8 +36,14 @@ export interface KafkaEventSourceConfig { const CONFIG_PREFIX_PUBLISHER = 'events.modules.kafka.kafkaConsumingEventPublisher'; -export const readConfig = (config: Config): KafkaEventSourceConfig => { - const kafkaConfig = config.getConfig(CONFIG_PREFIX_PUBLISHER); +export const readConfig = ( + config: Config, +): KafkaEventSourceConfig | undefined => { + const kafkaConfig = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + + if (!kafkaConfig) { + return undefined; + } const clientId = kafkaConfig.getString('clientId'); const brokers = kafkaConfig.getStringArray('brokers'); diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts index a458a39662..b524454ca0 100644 --- a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts @@ -43,6 +43,10 @@ export const eventsModuleKafkaConsumingEventPublisher = createBackendModule({ logger, }); + if (!kafka) { + return; + } + await kafka.start(); lifecycle.addShutdownHook(async () => await kafka.shutdown()); From c7065e4c266dcc2a89923e5ced221b708b6027b6 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 4 Jun 2025 12:30:30 +0200 Subject: [PATCH 14/15] chore(events): use readDurationFromConfig for all config durations Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/config.d.ts | 40 +++++----- .../events-backend-module-kafka/package.json | 3 +- .../src/publisher/config.test.ts | 80 +++++++++++++++---- .../src/publisher/config.ts | 77 +++++++++++++++--- yarn.lock | 1 + 5 files changed, 155 insertions(+), 46 deletions(-) diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts index 4c0d06c77e..c3400d43cd 100644 --- a/plugins/events-backend-module-kafka/config.d.ts +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -60,16 +60,16 @@ export interface Config { */ retry: { /** - * (Optional) Maximum wait time for a retry in milliseconds + * (Optional) Maximum wait time for a retry * Default: 30000 ms. */ - maxRetryTime: number; + maxRetryTime: HumanDuration | string; /** - * (Optional) Initial value used to calculate the retry in milliseconds (This is still randomized following the randomization factor) + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) * Default: 300 ms. */ - initialRetryTime: number; + initialRetryTime: HumanDuration | string; /** * (Optional) Randomization factor @@ -91,22 +91,22 @@ export interface Config { }; /** - * (Optional) Timeout in ms for authentication requests. + * (Optional) Timeout for authentication requests. * Default: 10000 ms. */ - authenticationTimeout: number; + authenticationTimeout: HumanDuration | string; /** - * (Optional) Time in milliseconds to wait for a successful connection. + * (Optional) Time to wait for a successful connection. * Default: 1000 ms. */ - connectionTimeout: number; + connectionTimeout: HumanDuration | string; /** - * (Optional) Time in milliseconds to wait for a successful request. + * (Optional) Time to wait for a successful request. * Default: 30000 ms. */ - requestTimeout: number; + requestTimeout: HumanDuration | string; /** * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. @@ -137,34 +137,34 @@ export interface Config { groupId: string; /** - * (Optional) Timeout in milliseconds used to detect failures. + * (Optional) Timeout used to detect failures. * The consumer sends periodic heartbeats to indicate its liveness to the broker. * If no heartbeats are received by the broker before the expiration of this session timeout, * then the broker will remove this consumer from the group and initiate a rebalance * Default: 30000 ms. */ - sessionTimeout: number; + sessionTimeout: HumanDuration | string; /** * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group * Default: 60000 ms. */ - rebalanceTimeout: number; + rebalanceTimeout: HumanDuration | string; /** - * (Optional) The expected time in milliseconds between heartbeats to the consumer coordinator. + * (Optional) The expected time between heartbeats to the consumer coordinator. * Heartbeats are used to ensure that the consumer's session stays active. * The value must be set lower than session timeout * Default: 3000 ms. */ - heartbeatInterval: number; + heartbeatInterval: HumanDuration | string; /** - * (Optional) The period of time in milliseconds after which we force a refresh of metadata + * (Optional) The period of time after which we force a refresh of metadata * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions * Default: 300000 ms (5 minutes). */ - metadataMaxAge: number; + metadataMaxAge: HumanDuration | string; /** * (Optional) The maximum amount of data per-partition the server will return. @@ -176,7 +176,7 @@ export interface Config { maxBytesPerPartition: number; /** - * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTimeInMs for more data to accumulate. + * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate. * Default: 1 */ minBytes: number; @@ -188,11 +188,11 @@ export interface Config { maxBytes: number; /** - * (Optional) The maximum amount of time in milliseconds the server will block before answering the fetch request + * (Optional) The maximum amount of time the server will block before answering the fetch request * if there isn’t sufficient data to immediately satisfy the requirement given by minBytes * Default: 5000 */ - maxWaitTimeInMs: number; + maxWaitTime: HumanDuration | string; }; }>; }; diff --git a/plugins/events-backend-module-kafka/package.json b/plugins/events-backend-module-kafka/package.json index 4b9594c4af..488ecc0a0e 100644 --- a/plugins/events-backend-module-kafka/package.json +++ b/plugins/events-backend-module-kafka/package.json @@ -39,7 +39,8 @@ "@backstage/config": "workspace:^", "@backstage/plugin-events-node": "workspace:^", "@backstage/types": "workspace:^", - "kafkajs": "^2.2.4" + "kafkajs": "^2.2.4", + "luxon": "^3.0.0" }, "devDependencies": { "@backstage/backend-test-utils": "workspace:^", diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts index 5b2dfb6f0c..fc7e3049b0 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts @@ -90,15 +90,15 @@ describe('readConfig', () => { password: 'password', }, retry: { - maxRetryTime: '20000', - initialRetryTime: '200', + maxRetryTime: { milliseconds: 20000 }, + initialRetryTime: { milliseconds: 200 }, factor: '0.4', multiplier: '4', retries: '10', }, - authenticationTimeout: 20000, - connectionTimeout: 1500, - requestTimeout: 20000, + authenticationTimeout: { milliseconds: 20000 }, + connectionTimeout: { milliseconds: 1500 }, + requestTimeout: { milliseconds: 20000 }, enforceRequestTimeout: false, topics: [ { @@ -106,14 +106,14 @@ describe('readConfig', () => { kafka: { topics: ['topic-A'], groupId: 'my-group', - sessionTimeout: 20000, - rebalanceTimeout: 50000, - heartbeatInterval: 2000, - metadataMaxAge: 400000, + sessionTimeout: { milliseconds: 20000 }, + rebalanceTimeout: { milliseconds: 50000 }, + heartbeatInterval: { milliseconds: 2000 }, + metadataMaxAge: { milliseconds: 400000 }, maxBytesPerPartition: 50000, minBytes: 2, maxBytes: 500000, - maxWaitTimeInMs: 4000, + maxWaitTime: { milliseconds: 4000 }, }, }, { @@ -151,11 +151,11 @@ describe('readConfig', () => { expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(20000); expect(publisherConfigs?.kafkaConfig.enforceRequestTimeout).toBeFalsy(); expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({ - maxRetryTime: '20000', - initialRetryTime: '200', - factor: '0.4', - multiplier: '4', - retries: '10', + maxRetryTime: 20000, + initialRetryTime: 200, + factor: 0.4, + multiplier: 4, + retries: 10, }); // Consumer configuration @@ -197,4 +197,54 @@ describe('readConfig', () => { publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxWaitTimeInMs, ).toBe(4000); }); + + it('should handle HumanDuration and string values for durations and timeouts', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + retry: { + maxRetryTime: { seconds: 1 }, + initialRetryTime: { minutes: 1 }, + factor: 0.4, + multiplier: 4, + retries: 10, + }, + authenticationTimeout: { hours: 1 }, + connectionTimeout: { days: 1 }, + topics: [], + requestTimeout: '1m', + }, + }, + }, + }, + }); + + const publisherConfigs = readConfig(config); + + expect(publisherConfigs).toBeDefined(); + + // Client configuration + expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events'); + expect(publisherConfigs?.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(publisherConfigs?.kafkaConfig.authenticationTimeout).toBe(3600000); + expect(publisherConfigs?.kafkaConfig.connectionTimeout).toBe(86400000); + expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(60000); + expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: 1000, + initialRetryTime: 60000, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + // Consumer configuration + expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(0); + }); }); diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts index aadf4a5166..aa5921a7f3 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -13,8 +13,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { Config } from '@backstage/config'; +import { Config, readDurationFromConfig } from '@backstage/config'; import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs'; +import { Duration } from 'luxon'; /** * @public @@ -36,6 +37,26 @@ export interface KafkaEventSourceConfig { const CONFIG_PREFIX_PUBLISHER = 'events.modules.kafka.kafkaConsumingEventPublisher'; +/** + * Reads an optional HumanDuration from the config and returns the value in milliseconds if the key is defined. + * + * @param config - The configuration object to read from. + * @param key - The key to look up in the configuration. + * @returns The duration in milliseconds, or undefined if the key is not defined. + */ +const readOptionalHumanDurationInMs = ( + config: Config, + key: string, +): number | undefined => { + const humanDuration = config.has(key) + ? readDurationFromConfig(config, { key }) + : undefined; + + if (!humanDuration) return undefined; + + return Duration.fromObject(humanDuration).as('milliseconds'); +}; + export const readConfig = ( config: Config, ): KafkaEventSourceConfig | undefined => { @@ -48,18 +69,39 @@ export const readConfig = ( const clientId = kafkaConfig.getString('clientId'); const brokers = kafkaConfig.getStringArray('brokers'); - const authenticationTimeout = kafkaConfig.getOptionalNumber( + const authenticationTimeout = readOptionalHumanDurationInMs( + kafkaConfig, 'authenticationTimeout', ); - const connectionTimeout = kafkaConfig.getOptionalNumber('connectionTimeout'); - const requestTimeout = kafkaConfig.getOptionalNumber('requestTimeout'); + + const connectionTimeout = readOptionalHumanDurationInMs( + kafkaConfig, + 'connectionTimeout', + ); + const requestTimeout = readOptionalHumanDurationInMs( + kafkaConfig, + 'requestTimeout', + ); const enforceRequestTimeout = kafkaConfig.getOptionalBoolean( 'enforceRequestTimeout', ); const ssl = kafkaConfig.getOptional('ssl') as KafkaConfig['ssl']; const sasl = kafkaConfig.getOptional('sasl') as KafkaConfig['sasl']; - const retry = kafkaConfig.getOptional('retry') as KafkaConfig['retry']; + + const retry: KafkaConfig['retry'] = { + maxRetryTime: readOptionalHumanDurationInMs( + kafkaConfig, + 'retry.maxRetryTime', + ), + initialRetryTime: readOptionalHumanDurationInMs( + kafkaConfig, + 'retry.initialRetryTime', + ), + factor: kafkaConfig.getOptionalNumber('retry.factor'), + multiplier: kafkaConfig.getOptionalNumber('retry.multiplier'), + retries: kafkaConfig.getOptionalNumber('retry.retries'), + }; const kafkaConsumerConfigs: KafkaConsumerConfig[] = kafkaConfig .getConfigArray('topics') @@ -68,16 +110,31 @@ export const readConfig = ( backstageTopic: topic.getString('topic'), consumerConfig: { groupId: topic.getString('kafka.groupId'), - sessionTimeout: topic.getOptionalNumber('kafka.sessionTimeout'), - rebalanceTimeout: topic.getOptionalNumber('kafka.rebalanceTimeout'), - heartbeatInterval: topic.getOptionalNumber('kafka.heartbeatInterval'), - metadataMaxAge: topic.getOptionalNumber('kafka.metadataMaxAge'), + sessionTimeout: readOptionalHumanDurationInMs( + topic, + 'kafka.sessionTimeout', + ), + rebalanceTimeout: readOptionalHumanDurationInMs( + topic, + 'kafka.rebalanceTimeout', + ), + heartbeatInterval: readOptionalHumanDurationInMs( + topic, + 'kafka.heartbeatInterval', + ), + metadataMaxAge: readOptionalHumanDurationInMs( + topic, + 'kafka.metadataMaxAge', + ), maxBytesPerPartition: topic.getOptionalNumber( 'kafka.maxBytesPerPartition', ), minBytes: topic.getOptionalNumber('kafka.minBytes'), maxBytes: topic.getOptionalNumber('kafka.maxBytes'), - maxWaitTimeInMs: topic.getOptionalNumber('kafka.maxWaitTimeInMs'), + maxWaitTimeInMs: readOptionalHumanDurationInMs( + topic, + 'kafka.maxWaitTime', + ), }, consumerSubscribeTopics: { topics: topic.getStringArray('kafka.topics'), diff --git a/yarn.lock b/yarn.lock index 11e3bd9ae4..6cadbc2dfd 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6647,6 +6647,7 @@ __metadata: "@backstage/plugin-events-node": "workspace:^" "@backstage/types": "workspace:^" kafkajs: "npm:^2.2.4" + luxon: "npm:^3.0.0" languageName: unknown linkType: soft From 82384908685c7300e989330713854f896eb7465e Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 4 Jun 2025 12:40:29 +0200 Subject: [PATCH 15/15] chore(events): add HumanDuration import to config.d.ts Signed-off-by: Jonas Beck --- plugins/events-backend-module-kafka/config.d.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts index c3400d43cd..49c417098c 100644 --- a/plugins/events-backend-module-kafka/config.d.ts +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -13,6 +13,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + +import { HumanDuration } from '@backstage/types'; + export interface Config { events?: { modules?: {