From 805c48d23d742e0a6bb735f4c6b316b8eb8a3dc7 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Mon, 8 Dec 2025 12:37:02 +0100 Subject: [PATCH 1/5] feat(events): restructure kafka module and add publisher Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/config.d.ts | 495 ++++++++++++------ .../KafkaConsumingEventPublisher.test.ts | 163 ++++++ .../KafkaConsumingEventPublisher.ts | 109 ++++ .../config.test.ts | 275 ++++++++++ .../KafkaConsumingEventPublisher/config.ts | 91 ++++ .../src/KafkaConsumingEventPublisher/index.ts | 16 + .../module.test.ts | 122 +++++ .../module.ts} | 20 +- .../KafkaPublishingEventConsumer.test.ts | 106 ++++ .../KafkaPublishingEventConsumer.ts | 115 ++++ .../config.test.ts | 360 +++++++++++++ .../KafkaPublishingEventConsumer/config.ts | 82 +++ .../src/KafkaPublishingEventConsumer/index.ts | 16 + .../module.test.ts | 122 +++++ .../KafkaPublishingEventConsumer/module.ts | 56 ++ .../events-backend-module-kafka/src/index.ts | 16 +- .../src/publisher/KafkaConsumerClient.test.ts | 132 ----- .../src/publisher/KafkaConsumerClient.ts | 77 --- .../KafkaConsumingEventPublisher.test.ts | 92 ---- .../publisher/KafkaConsumingEventPublisher.ts | 109 ---- .../src/publisher/config.test.ts | 250 --------- .../src/publisher/config.ts | 153 ------ ...ModuleKafkaConsumingEventPublisher.test.ts | 85 --- .../LoggerServiceAdapter.ts | 0 .../src/utils/config.test.ts | 235 +++++++++ .../src/utils/config.ts | 84 +++ .../src/utils/kafkaTransformers.test.ts | 137 +++++ .../src/utils/kafkaTransformers.ts | 49 ++ 28 files changed, 2504 insertions(+), 1063 deletions(-) create mode 100644 plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.test.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/index.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.test.ts rename plugins/events-backend-module-kafka/src/{service/eventsModuleKafkaConsumingEventPublisher.ts => KafkaConsumingEventPublisher/module.ts} (69%) create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.test.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.test.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/index.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.test.ts create mode 100644 plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.ts delete mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts delete mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts delete mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts delete mode 100644 plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts delete mode 100644 plugins/events-backend-module-kafka/src/publisher/config.test.ts delete mode 100644 plugins/events-backend-module-kafka/src/publisher/config.ts delete mode 100644 plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts rename plugins/events-backend-module-kafka/src/{publisher => utils}/LoggerServiceAdapter.ts (100%) create mode 100644 plugins/events-backend-module-kafka/src/utils/config.test.ts create mode 100644 plugins/events-backend-module-kafka/src/utils/config.ts create mode 100644 plugins/events-backend-module-kafka/src/utils/kafkaTransformers.test.ts create mode 100644 plugins/events-backend-module-kafka/src/utils/kafkaTransformers.ts diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts index 49c417098c..b3698a8999 100644 --- a/plugins/events-backend-module-kafka/config.d.ts +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - import { HumanDuration } from '@backstage/types'; export interface Config { @@ -25,179 +24,373 @@ export interface Config { kafka?: { /** * Configuration for KafkaConsumingEventPublisher + * + * Supports multiple named instances as a record where each key is a unique name + * for the Kafka consumer configuration. */ 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: { + [name: string]: { /** - * (Optional) Maximum wait time for a retry - * Default: 30000 ms. + * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. */ - maxRetryTime: HumanDuration | string; + 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) Initial value used to calculate the retry (This is still randomized following the randomization factor) - * Default: 300 ms. + * Optional retry connection parameters. */ - initialRetryTime: HumanDuration | string; - - /** - * (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 for authentication requests. - * Default: 10000 ms. - */ - authenticationTimeout: HumanDuration | string; - - /** - * (Optional) Time to wait for a successful connection. - * Default: 1000 ms. - */ - connectionTimeout: HumanDuration | string; - - /** - * (Optional) Time to wait for a successful request. - * Default: 30000 ms. - */ - requestTimeout: HumanDuration | string; - - /** - * (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: { + retry?: { /** - * (Required) The Kafka topics to subscribe to - */ - topics: string[]; - /** - * (Required) The GroupId to be used by the topic consumers - */ - groupId: string; - - /** - * (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 + * (Optional) Maximum wait time for a retry * Default: 30000 ms. */ - sessionTimeout: HumanDuration | string; + maxRetryTime?: HumanDuration | string; /** - * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group - * Default: 60000 ms. + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. */ - rebalanceTimeout: HumanDuration | string; + initialRetryTime?: HumanDuration | string; /** - * (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. + * (Optional) Randomization factor + * Default: 0.2. */ - heartbeatInterval: HumanDuration | string; + factor?: number; /** - * (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). + * (Optional) Exponential factor + * Default: 2. */ - metadataMaxAge: HumanDuration | string; + multiplier?: 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) + * (Optional) Max number of retries per call + * Default: 5. */ - maxBytesPerPartition: number; - - /** - * (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; - - /** - * (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 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 - */ - maxWaitTime: HumanDuration | string; + retries?: number; }; - }>; + + /** + * (Optional) Timeout for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout?: HumanDuration | string; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout?: boolean; + + /** + * Contains an object per topic for which a 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 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?: HumanDuration | string; + + /** + * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group + * Default: 60000 ms. + */ + rebalanceTimeout?: HumanDuration | string; + + /** + * (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?: HumanDuration | string; + + /** + * (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?: HumanDuration | string; + + /** + * (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 maxWaitTime 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 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 + */ + maxWaitTime?: HumanDuration | string; + }; + }>; + }; + }; + + /** + * Configuration for KafkaPublishingEventConsumer + * + * Supports multiple named instances as a record where each key is a unique name + * for the Kafka producer configuration. + */ + kafkaPublishingEventConsumer?: { + [name: string]: { + /** + * (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 + * Default: 30000 ms. + */ + maxRetryTime?: HumanDuration | string; + + /** + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. + */ + initialRetryTime?: HumanDuration | string; + + /** + * (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 for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout?: HumanDuration | string; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout?: boolean; + + /** + * Contains an object per topic for which a Kafka queue + * should be used as destination for events. + */ + topics: Array<{ + /** + * (Required) The Backstage topic to consume from + */ + topic: string; + /** + * (Required) KafkaProducer-related configuration. + */ + kafka: { + /** + * (Required) The Kafka topic to publish to + */ + topic: string; + + /** + * (Optional) Allow topic creation when querying metadata for non-existent topics. + * Default: true + */ + allowAutoTopicCreation?: boolean; + + /** + * (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?: HumanDuration | string; + + /** + * (Optional) The maximum amount of time in ms that the transaction coordinator will wait for a transaction status update + * from the producer before proactively aborting the ongoing transaction. + * If this value is larger than the `transaction.max.timeout.ms`` setting in the broker, the request will fail with a `InvalidTransactionTimeout` error + * Default: 60000 ms. + */ + transactionTimeout?: HumanDuration | string; + + /** + * (Optional) Experimental. If enabled producer will ensure each message is written exactly once. Acks must be set to -1 ("all"). + * Retries will default to MAX_SAFE_INTEGER. + * Default: false. + */ + idempotent?: boolean; + + /** + * (Optional) Max number of requests that may be in progress at any time. If falsey then no limit. + * Default: null. + */ + maxInFlightRequests?: number; + + /** + * Optional retry connection parameters. + */ + retry?: { + /** + * (Optional) Maximum wait time for a retry + * Default: 30000 ms. + */ + maxRetryTime?: HumanDuration | string; + + /** + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. + */ + initialRetryTime?: HumanDuration | string; + + /** + * (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; + }; + }; + }>; + }; }; }; }; diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.test.ts new file mode 100644 index 0000000000..98781936b0 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.test.ts @@ -0,0 +1,163 @@ +/* + * 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 { ConfigReader } from '@backstage/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; + + jest.mocked(Kafka).mockImplementation(() => mockKafkaClient); + + const mockConfig = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: [ + { + topic: 'backstage-topic', + kafka: { + topics: ['test-topic'], + groupId: 'test-group', + }, + }, + ], + }, + }, + }, + }, + }, + }); + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('should create instances from config', () => { + const consumers = KafkaConsumingEventPublisher.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(consumers).toBeInstanceOf(Array); + expect(consumers).toHaveLength(1); + expect(consumers[0]).toBeInstanceOf(KafkaConsumingEventPublisher); + }); + + it('should return empty array when no config', () => { + const consumers = KafkaConsumingEventPublisher.fromConfig({ + config: new ConfigReader({}), + events: mockEvents, + logger: mockLogger, + }); + + expect(consumers).toEqual([]); + }); + + it('should start all consumers', async () => { + const consumers = KafkaConsumingEventPublisher.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(consumers).toHaveLength(1); + + await consumers[0].start(); + + expect(mockConsumer.connect).toHaveBeenCalled(); + expect(mockConsumer.subscribe).toHaveBeenCalledWith({ + topics: ['test-topic'], + }); + expect(mockConsumer.run).toHaveBeenCalled(); + }); + + it('should shutdown all consumers', async () => { + const consumers = KafkaConsumingEventPublisher.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(consumers).toHaveLength(1); + + await consumers[0].shutdown(); + + expect(mockConsumer.disconnect).toHaveBeenCalled(); + }); + + it('should handle multiple consumer configs', () => { + const multiConsumerConfig = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092'], + topics: [ + { + topic: 'topic1', + kafka: { + topics: ['kafka-topic-1'], + groupId: 'group1', + }, + }, + { + topic: 'topic2', + kafka: { + topics: ['kafka-topic-2'], + groupId: 'group2', + }, + }, + ], + }, + }, + }, + }, + }, + }); + + const consumers = KafkaConsumingEventPublisher.fromConfig({ + config: multiConsumerConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(consumers).toHaveLength(1); + expect(mockKafkaClient.consumer).toHaveBeenCalledTimes(2); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts new file mode 100644 index 0000000000..41744abff2 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts @@ -0,0 +1,109 @@ +/* + * 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 { EventsService } from '@backstage/plugin-events-node'; +import { Consumer, Kafka } from 'kafkajs'; +import { + KafkaConsumerConfig, + KafkaConsumingEventPublisherConfig, + readConsumerConfig, +} from './config'; +import { Config } from '@backstage/config'; +import { loggerServiceAdapter } from '../utils/LoggerServiceAdapter'; +import { convertHeadersToMetadata } from '../utils/kafkaTransformers'; + +type KafkaConsumer = { + consumer: Consumer; + config: KafkaConsumerConfig; +}; + +/** + * 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. + */ +export class KafkaConsumingEventPublisher { + private readonly kafkaConsumers: KafkaConsumer[]; + private readonly logger: LoggerService; + + static fromConfig(env: { + config: Config; + events: EventsService; + logger: LoggerService; + }): KafkaConsumingEventPublisher[] { + const configs = readConsumerConfig(env.config); + + return configs.map( + kafkaConfig => + new KafkaConsumingEventPublisher(env.logger, env.events, kafkaConfig), + ); + } + + private constructor( + logger: LoggerService, + private readonly events: EventsService, + config: KafkaConsumingEventPublisherConfig, + ) { + this.logger = logger.child({ + class: KafkaConsumingEventPublisher.prototype.constructor.name, + instance: config.instance, + }); + + const kafka = new Kafka({ + ...config.kafkaConfig, + logCreator: loggerServiceAdapter(this.logger), + }); + + this.kafkaConsumers = config.kafkaConsumerConfigs.map(consumerConfig => ({ + consumer: kafka.consumer(consumerConfig.consumerConfig), + config: consumerConfig, + })); + } + + async start(): Promise { + await Promise.all( + this.kafkaConsumers.map(async ({ consumer, config }) => { + const consumerLogger = this.logger.child({ + id: `events.kafka.publisher:${config.backstageTopic}`, + groupId: config.consumerConfig.groupId, + kafkaTopics: config.consumerSubscribeTopics.topics.toString(), + backstageTopic: config.backstageTopic, + }); + try { + await consumer.connect(); + await consumer.subscribe(config.consumerSubscribeTopics); + + await consumer.run({ + eachMessage: async ({ message }) => { + this.events.publish({ + topic: config.backstageTopic, + eventPayload: JSON.parse(message.value?.toString()!), + metadata: convertHeadersToMetadata(message.headers), + }); + }, + }); + } catch (error: any) { + consumerLogger.error('Kafka consumer connection failed', error); + } + }), + ); + } + + async shutdown(): Promise { + await Promise.all( + this.kafkaConsumers.map(({ consumer }) => consumer.disconnect()), + ); + } +} diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts new file mode 100644 index 0000000000..b61ee9332e --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts @@ -0,0 +1,275 @@ +/* + * 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 { readConsumerConfig } from './config'; + +describe('readConsumerConfig', () => { + it('not configured', () => { + const publisherConfigs = readConsumerConfig(new ConfigReader({})); + + expect(publisherConfigs).toEqual([]); + }); + + it('only required fields configured', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + dev: { + 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 = readConsumerConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const devConfig = publisherConfigs[0]; + expect(devConfig.instance).toBe('dev'); + expect(devConfig.kafkaConsumerConfigs.length).toBe(2); + + expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(devConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + + expect(devConfig.kafkaConsumerConfigs).toEqual([ + { + backstageTopic: 'fake1', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-A'], + }, + }, + { + backstageTopic: 'fake2', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-B'], + }, + }, + ]); + }); + + it('all fields configured', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + ssl: true, + sasl: { + mechanism: 'plain', + username: 'username', + password: 'password', + }, + retry: { + maxRetryTime: { milliseconds: 20000 }, + initialRetryTime: { milliseconds: 200 }, + factor: '0.4', + multiplier: '4', + retries: '10', + }, + authenticationTimeout: { milliseconds: 20000 }, + connectionTimeout: { milliseconds: 1500 }, + requestTimeout: { milliseconds: 20000 }, + enforceRequestTimeout: false, + topics: [ + { + topic: 'fake1', + kafka: { + topics: ['topic-A'], + groupId: 'my-group', + sessionTimeout: { milliseconds: 20000 }, + rebalanceTimeout: { milliseconds: 50000 }, + heartbeatInterval: { milliseconds: 2000 }, + metadataMaxAge: { milliseconds: 400000 }, + maxBytesPerPartition: 50000, + minBytes: 2, + maxBytes: 500000, + maxWaitTime: { milliseconds: 4000 }, + }, + }, + { + topic: 'fake2', + kafka: { + topics: ['topic-B'], + groupId: 'my-group', + }, + }, + ], + }, + }, + }, + }, + }, + }); + + const publisherConfigs = readConsumerConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const devConfig = publisherConfigs[0]; + expect(devConfig.instance).toBe('dev'); + + // Client configuration + expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(devConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(devConfig.kafkaConfig.ssl).toBeTruthy(); + expect(devConfig.kafkaConfig.sasl).toStrictEqual({ + mechanism: 'plain', + username: 'username', + password: 'password', + }); + expect(devConfig.kafkaConfig.authenticationTimeout).toBe(20000); + expect(devConfig.kafkaConfig.connectionTimeout).toBe(1500); + expect(devConfig.kafkaConfig.requestTimeout).toBe(20000); + expect(devConfig.kafkaConfig.enforceRequestTimeout).toBeFalsy(); + expect(devConfig.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: 20000, + initialRetryTime: 200, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + // Consumer configuration + expect(devConfig.kafkaConsumerConfigs.length).toBe(2); + + expect(devConfig.kafkaConsumerConfigs).toEqual([ + { + backstageTopic: 'fake1', + consumerConfig: { + groupId: 'my-group', + sessionTimeout: 20000, + rebalanceTimeout: 50000, + heartbeatInterval: 2000, + metadataMaxAge: 400000, + maxBytesPerPartition: 50000, + minBytes: 2, + maxBytes: 500000, + maxWaitTimeInMs: 4000, + }, + consumerSubscribeTopics: { + topics: ['topic-A'], + }, + }, + { + backstageTopic: 'fake2', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-B'], + }, + }, + ]); + }); + + it('should handle HumanDuration and string values for durations and timeouts', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + dev: { + 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 = readConsumerConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const devConfig = publisherConfigs[0]; + expect(devConfig.instance).toBe('dev'); + + // Client configuration + expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(devConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(devConfig.kafkaConfig.authenticationTimeout).toBe(3600000); + expect(devConfig.kafkaConfig.connectionTimeout).toBe(86400000); + expect(devConfig.kafkaConfig.requestTimeout).toBe(60000); + expect(devConfig.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: 1000, + initialRetryTime: 60000, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + // Consumer configuration + expect(devConfig.kafkaConsumerConfigs.length).toBe(0); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts new file mode 100644 index 0000000000..dc9b477770 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts @@ -0,0 +1,91 @@ +/* + * 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'; +import { + readKafkaConfig, + readOptionalHumanDurationInMs, +} from '../utils/config'; + +export interface KafkaConsumerConfig { + backstageTopic: string; + consumerConfig: ConsumerConfig; + consumerSubscribeTopics: ConsumerSubscribeTopics; +} + +export interface KafkaConsumingEventPublisherConfig { + instance: string; + kafkaConfig: KafkaConfig; + kafkaConsumerConfigs: KafkaConsumerConfig[]; +} + +const CONFIG_PREFIX_PUBLISHER = + 'events.modules.kafka.kafkaConsumingEventPublisher'; + +export const readConsumerConfig = ( + config: Config, +): KafkaConsumingEventPublisherConfig[] => { + const publishers = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + + return ( + publishers?.keys()?.map(publisherKey => { + const publisherConfig = publishers.getConfig(publisherKey); + + return { + instance: publisherKey, + kafkaConfig: readKafkaConfig(publisherConfig), + kafkaConsumerConfigs: publisherConfig + .getConfigArray('topics') + .map(topicConfig => { + return { + backstageTopic: topicConfig.getString('topic'), + consumerConfig: { + groupId: topicConfig.getString('kafka.groupId'), + sessionTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.sessionTimeout', + ), + rebalanceTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.rebalanceTimeout', + ), + heartbeatInterval: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.heartbeatInterval', + ), + metadataMaxAge: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.metadataMaxAge', + ), + maxBytesPerPartition: topicConfig.getOptionalNumber( + 'kafka.maxBytesPerPartition', + ), + minBytes: topicConfig.getOptionalNumber('kafka.minBytes'), + maxBytes: topicConfig.getOptionalNumber('kafka.maxBytes'), + maxWaitTimeInMs: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.maxWaitTime', + ), + }, + consumerSubscribeTopics: { + topics: topicConfig.getStringArray('kafka.topics'), + }, + }; + }), + }; + }) ?? [] + ); +}; diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/index.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/index.ts new file mode 100644 index 0000000000..94d36a81e0 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/index.ts @@ -0,0 +1,16 @@ +/* + * 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 { eventsModuleKafkaConsumingEventPublisher } from './module'; diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.test.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.test.ts new file mode 100644 index 0000000000..19ac969800 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.test.ts @@ -0,0 +1,122 @@ +/* + * 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 './module'; +import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; + +jest.mock('./KafkaConsumingEventPublisher'); + +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 mockKafkaConsumingEventPublisher = { + start: jest.fn(), + shutdown: jest.fn(), + } as unknown as KafkaConsumingEventPublisher; + + jest + .mocked(KafkaConsumingEventPublisher.fromConfig) + .mockReturnValue([mockKafkaConsumingEventPublisher]); + + await startTestBackend({ + features: [ + eventsServiceFactory, + eventsModuleKafkaConsumingEventPublisher, + mockServices.rootConfig.factory({ + data: { + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + dev: { + 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(mockKafkaConsumingEventPublisher.start).toHaveBeenCalled(); + + // Verify that the shutdown hook was registered + expect(mockKafkaConsumingEventPublisher.shutdown).not.toHaveBeenCalled(); + }); + + it('should handle empty configuration gracefully', async () => { + const events = new TestEventsService(); + const eventsServiceFactory = createServiceFactory({ + service: eventsServiceRef, + deps: {}, + async factory({}) { + return events; + }, + }); + + jest.mocked(KafkaConsumingEventPublisher.fromConfig).mockReturnValue([]); + + await startTestBackend({ + features: [ + eventsServiceFactory, + eventsModuleKafkaConsumingEventPublisher, + mockServices.rootConfig.factory({ + data: { + events: { + modules: { + kafka: { + // No kafkaConsumingEventPublisher config + }, + }, + }, + }, + }), + ], + }); + + // Verify that fromConfig was called but returned empty array + expect(KafkaConsumingEventPublisher.fromConfig).toHaveBeenCalled(); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.ts similarity index 69% rename from plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts rename to plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.ts index b524454ca0..8ad3396c76 100644 --- a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/module.ts @@ -17,11 +17,11 @@ import { coreServices, createBackendModule, } from '@backstage/backend-plugin-api'; -import { KafkaConsumerClient } from '../publisher/KafkaConsumerClient'; import { eventsServiceRef } from '@backstage/plugin-events-node'; +import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; /** - * Kafka module for the Events plugin. + * Reads messages off of Kafka topics and forwards them into the Backstage events system. * * @public */ @@ -34,22 +34,22 @@ export const eventsModuleKafkaConsumingEventPublisher = createBackendModule({ config: coreServices.rootConfig, events: eventsServiceRef, logger: coreServices.logger, - lifecycle: coreServices.lifecycle, + lifecycle: coreServices.rootLifecycle, }, async init({ config, logger, events, lifecycle }) { - const kafka = KafkaConsumerClient.fromConfig({ + const consumers = KafkaConsumingEventPublisher.fromConfig({ config, events, logger, }); - if (!kafka) { - return; - } + lifecycle.addStartupHook(async () => { + await Promise.all(consumers.map(consumer => consumer.start())); + }); - await kafka.start(); - - lifecycle.addShutdownHook(async () => await kafka.shutdown()); + lifecycle.addShutdownHook(async () => { + await Promise.all(consumers.map(consumer => consumer.shutdown())); + }); }, }); }, diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.test.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.test.ts new file mode 100644 index 0000000000..6379605901 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.test.ts @@ -0,0 +1,106 @@ +/* + * 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 { KafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer'; +import { Kafka } from 'kafkajs'; +import { mockServices } from '@backstage/backend-test-utils'; +import { ConfigReader } from '@backstage/config'; + +jest.mock('kafkajs'); + +describe('KafkaPublishingEventConsumer', () => { + const mockLogger = mockServices.logger.mock(); + const mockEvents = mockServices.events.mock(); + + const mockProducer = { + connect: jest.fn(), + disconnect: jest.fn(), + send: jest.fn(), + }; + + const mockKafkaClient = { + producer: jest.fn().mockReturnValue(mockProducer), + } as unknown as Kafka; + + jest.mocked(Kafka).mockImplementation(() => mockKafkaClient); + + const mockConfig = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaPublishingEventConsumer: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092'], + topics: [ + { + topic: 'backstage-topic', + kafka: { + topic: 'kafka-topic', + allowAutoTopicCreation: true, + }, + }, + ], + }, + }, + }, + }, + }, + }); + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('should create instances from config', () => { + const consumers = KafkaPublishingEventConsumer.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + expect(consumers).toHaveLength(1); + expect(consumers[0]).toBeInstanceOf(KafkaPublishingEventConsumer); + }); + + it('should start the consumer and subscribe to events', async () => { + const consumers = KafkaPublishingEventConsumer.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + await consumers[0].start(); + + expect(mockProducer.connect).toHaveBeenCalled(); + expect(mockEvents.subscribe).toHaveBeenCalledWith({ + id: 'kafka:publisher:backstage-topic', + topics: ['backstage-topic'], + onEvent: expect.any(Function), + }); + }); + + it('should shutdown the producer', async () => { + const consumers = KafkaPublishingEventConsumer.fromConfig({ + config: mockConfig, + events: mockEvents, + logger: mockLogger, + }); + + await consumers[0].shutdown(); + + expect(mockProducer.disconnect).toHaveBeenCalled(); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.ts new file mode 100644 index 0000000000..f9443f2d69 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/KafkaPublishingEventConsumer.ts @@ -0,0 +1,115 @@ +/* + * 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 { Kafka, Producer } from 'kafkajs'; +import { + KafkaPublisherConfig, + KafkaPublishingEventConsumerConfig, + readPublisherConfig, +} from './config'; +import { Config } from '@backstage/config'; +import { loggerServiceAdapter } from '../utils/LoggerServiceAdapter'; +import { payloadToBuffer } from '../utils/kafkaTransformers'; + +type KafkaPublisher = { + producer: Producer; + config: KafkaPublisherConfig; +}; + +/** + * This class subscribes to Backstage internal events and publishes them to Kafka topics. + * The internal event payload will be serialized and sent to the configured Kafka topic. + */ +export class KafkaPublishingEventConsumer { + private readonly kafkaPublishers: KafkaPublisher[]; + private readonly logger: LoggerService; + + static fromConfig(env: { + config: Config; + events: EventsService; + logger: LoggerService; + }): KafkaPublishingEventConsumer[] { + const configs = readPublisherConfig(env.config); + + return configs.map( + kafkaConfig => + new KafkaPublishingEventConsumer(env.logger, env.events, kafkaConfig), + ); + } + + private constructor( + logger: LoggerService, + private readonly events: EventsService, + config: KafkaPublishingEventConsumerConfig, + ) { + this.logger = logger.child({ + class: KafkaPublishingEventConsumer.prototype.constructor.name, + instance: config.instance, + }); + + const kafka = new Kafka({ + ...config.kafkaConfig, + logCreator: loggerServiceAdapter(this.logger), + }); + + this.kafkaPublishers = config.kafkaPublisherConfigs.map( + publisherConfig => ({ + producer: kafka.producer(publisherConfig.producerConfig), + config: publisherConfig, + }), + ); + } + + async start(): Promise { + await Promise.all( + this.kafkaPublishers.map(async ({ producer, config }) => { + try { + await producer.connect(); + + this.events.subscribe({ + id: `kafka:publisher:${config.backstageTopic}`, + topics: [config.backstageTopic], + onEvent: async (params: EventParams) => { + await producer.send({ + topic: config.kafkaTopic, + messages: [ + { + value: payloadToBuffer(params.eventPayload), + }, + ], + }); + }, + }); + this.logger.info( + `Subscribed to EventService, publishing events to external topic: ${config.backstageTopic}`, + ); + } catch (error: any) { + this.logger.error( + `Kafka producer connection failed for topic ${config.backstageTopic}`, + error, + ); + } + }), + ); + } + + async shutdown(): Promise { + await Promise.all( + this.kafkaPublishers.map(({ producer }) => producer.disconnect()), + ); + } +} diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.test.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.test.ts new file mode 100644 index 0000000000..bd4d622238 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.test.ts @@ -0,0 +1,360 @@ +/* + * 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 { readPublisherConfig } from './config'; + +describe('readPublisherConfig', () => { + it('not configured', () => { + const publisherConfigs = readPublisherConfig(new ConfigReader({})); + + expect(publisherConfigs).toEqual([]); + }); + + it('only required fields configured', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaPublishingEventConsumer: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: [ + { + topic: 'fake1', + kafka: { + topic: 'topic-A', + }, + }, + { + topic: 'fake2', + kafka: { + topic: 'topic-B', + }, + }, + ], + }, + }, + }, + }, + }, + }); + + const publisherConfigs = readPublisherConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const devConfig = publisherConfigs[0]; + expect(devConfig.instance).toBe('dev'); + expect(devConfig.kafkaPublisherConfigs.length).toBe(2); + + expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(devConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + + expect(devConfig.kafkaPublisherConfigs).toEqual([ + { + backstageTopic: 'fake1', + kafkaTopic: 'topic-A', + producerConfig: { + allowAutoTopicCreation: undefined, + metadataMaxAge: undefined, + transactionTimeout: undefined, + idempotent: undefined, + maxInFlightRequests: undefined, + retry: {}, + }, + }, + { + backstageTopic: 'fake2', + kafkaTopic: 'topic-B', + producerConfig: { + allowAutoTopicCreation: undefined, + metadataMaxAge: undefined, + transactionTimeout: undefined, + idempotent: undefined, + maxInFlightRequests: undefined, + retry: {}, + }, + }, + ]); + }); + + it('all fields configured', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaPublishingEventConsumer: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + ssl: true, + sasl: { + mechanism: 'plain', + username: 'username', + password: 'password', + }, + retry: { + maxRetryTime: { milliseconds: 20000 }, + initialRetryTime: { milliseconds: 200 }, + factor: '0.4', + multiplier: '4', + retries: '10', + }, + authenticationTimeout: { milliseconds: 20000 }, + connectionTimeout: { milliseconds: 1500 }, + requestTimeout: { milliseconds: 20000 }, + enforceRequestTimeout: false, + topics: [ + { + topic: 'fake1', + kafka: { + topic: 'topic-A', + allowAutoTopicCreation: true, + metadataMaxAge: { milliseconds: 400000 }, + transactionTimeout: { milliseconds: 30000 }, + idempotent: true, + maxInFlightRequests: 5, + retry: { + maxRetryTime: { milliseconds: 15000 }, + initialRetryTime: { milliseconds: 100 }, + factor: '0.2', + multiplier: '2', + retries: '5', + }, + }, + }, + { + topic: 'fake2', + kafka: { + topic: 'topic-B', + }, + }, + ], + }, + }, + }, + }, + }, + }); + + const publisherConfigs = readPublisherConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const devConfig = publisherConfigs[0]; + expect(devConfig.instance).toBe('dev'); + + // Client configuration + expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(devConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(devConfig.kafkaConfig.ssl).toBeTruthy(); + expect(devConfig.kafkaConfig.sasl).toStrictEqual({ + mechanism: 'plain', + username: 'username', + password: 'password', + }); + expect(devConfig.kafkaConfig.authenticationTimeout).toBe(20000); + expect(devConfig.kafkaConfig.connectionTimeout).toBe(1500); + expect(devConfig.kafkaConfig.requestTimeout).toBe(20000); + expect(devConfig.kafkaConfig.enforceRequestTimeout).toBeFalsy(); + expect(devConfig.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: 20000, + initialRetryTime: 200, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + // Publisher configuration + expect(devConfig.kafkaPublisherConfigs.length).toBe(2); + + expect(devConfig.kafkaPublisherConfigs).toEqual([ + { + backstageTopic: 'fake1', + kafkaTopic: 'topic-A', + producerConfig: { + allowAutoTopicCreation: true, + metadataMaxAge: 400000, + transactionTimeout: 30000, + idempotent: true, + maxInFlightRequests: 5, + retry: { + maxRetryTime: 15000, + initialRetryTime: 100, + factor: 0.2, + multiplier: 2, + retries: 5, + }, + }, + }, + { + backstageTopic: 'fake2', + kafkaTopic: 'topic-B', + producerConfig: { + allowAutoTopicCreation: undefined, + metadataMaxAge: undefined, + transactionTimeout: undefined, + idempotent: undefined, + maxInFlightRequests: undefined, + retry: {}, + }, + }, + ]); + }); + + it('should handle HumanDuration and string values for durations and timeouts', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaPublishingEventConsumer: { + dev: { + 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: [ + { + topic: 'fake1', + kafka: { + topic: 'topic-A', + metadataMaxAge: { seconds: 300 }, + transactionTimeout: '30s', + }, + }, + ], + requestTimeout: '1m', + }, + }, + }, + }, + }, + }); + + const publisherConfigs = readPublisherConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const devConfig = publisherConfigs[0]; + expect(devConfig.instance).toBe('dev'); + + // Client configuration + expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(devConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(devConfig.kafkaConfig.authenticationTimeout).toBe(3600000); + expect(devConfig.kafkaConfig.connectionTimeout).toBe(86400000); + expect(devConfig.kafkaConfig.requestTimeout).toBe(60000); + expect(devConfig.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: 1000, + initialRetryTime: 60000, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + // Publisher configuration + expect(devConfig.kafkaPublisherConfigs.length).toBe(1); + expect( + devConfig.kafkaPublisherConfigs[0].producerConfig.metadataMaxAge, + ).toBe(300000); + expect( + devConfig.kafkaPublisherConfigs[0].producerConfig.transactionTimeout, + ).toBe(30000); + }); + + it('should handle multiple instances', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaPublishingEventConsumer: { + dev: { + clientId: 'backstage-dev', + brokers: ['kafka-dev:9092'], + topics: [ + { + topic: 'dev-topic', + kafka: { + topic: 'kafka-dev-topic', + }, + }, + ], + }, + prod: { + clientId: 'backstage-prod', + brokers: ['kafka-prod1:9092', 'kafka-prod2:9092'], + topics: [ + { + topic: 'prod-topic', + kafka: { + topic: 'kafka-prod-topic', + }, + }, + ], + }, + }, + }, + }, + }, + }); + + const publisherConfigs = readPublisherConfig(config); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(2); + + const devConfig = publisherConfigs.find(c => c.instance === 'dev')!; + expect(devConfig.kafkaConfig.clientId).toBe('backstage-dev'); + expect(devConfig.kafkaConfig.brokers).toEqual(['kafka-dev:9092']); + expect(devConfig.kafkaPublisherConfigs).toHaveLength(1); + expect(devConfig.kafkaPublisherConfigs[0].backstageTopic).toBe('dev-topic'); + + const prodConfig = publisherConfigs.find(c => c.instance === 'prod')!; + expect(prodConfig.kafkaConfig.clientId).toBe('backstage-prod'); + expect(prodConfig.kafkaConfig.brokers).toEqual([ + 'kafka-prod1:9092', + 'kafka-prod2:9092', + ]); + expect(prodConfig.kafkaPublisherConfigs).toHaveLength(1); + expect(prodConfig.kafkaPublisherConfigs[0].backstageTopic).toBe( + 'prod-topic', + ); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.ts new file mode 100644 index 0000000000..8439e24405 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/config.ts @@ -0,0 +1,82 @@ +/* + * 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 { + readKafkaConfig, + readOptionalHumanDurationInMs, + readRetryConfig, +} from '../utils/config'; +import { KafkaConfig, ProducerConfig } from 'kafkajs'; + +export interface KafkaPublisherConfig { + backstageTopic: string; + kafkaTopic: string; + producerConfig: ProducerConfig; +} + +export interface KafkaPublishingEventConsumerConfig { + instance: string; + kafkaConfig: KafkaConfig; + kafkaPublisherConfigs: KafkaPublisherConfig[]; +} + +const CONFIG_PREFIX_PUBLISHER = + 'events.modules.kafka.kafkaPublishingEventConsumer'; + +export const readPublisherConfig = ( + config: Config, +): KafkaPublishingEventConsumerConfig[] => { + const publishers = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + + return ( + publishers?.keys()?.map(publisherKey => { + const publisherConfig = publishers.getConfig(publisherKey); + + return { + instance: publisherKey, + kafkaConfig: readKafkaConfig(publisherConfig), + kafkaPublisherConfigs: publisherConfig + .getConfigArray('topics') + .map(topicConfig => { + return { + backstageTopic: topicConfig.getString('topic'), + kafkaTopic: topicConfig.getString('kafka.topic'), + producerConfig: { + allowAutoTopicCreation: topicConfig.getOptionalBoolean( + 'kafka.allowAutoTopicCreation', + ), + metadataMaxAge: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.metadataMaxAge', + ), + transactionTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.transactionTimeout', + ), + idempotent: topicConfig.getOptionalBoolean('kafka.idempotent'), + maxInFlightRequests: topicConfig.getOptionalNumber( + 'kafka.maxInFlightRequests', + ), + retry: readRetryConfig( + topicConfig.getOptionalConfig('kafka.retry'), + ), + }, + }; + }), + }; + }) ?? [] + ); +}; diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/index.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/index.ts new file mode 100644 index 0000000000..42fadd2a5a --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/index.ts @@ -0,0 +1,16 @@ +/* + * 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 { eventsModuleKafkaPublishingEventConsumer } from './module'; diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.test.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.test.ts new file mode 100644 index 0000000000..39c8268300 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.test.ts @@ -0,0 +1,122 @@ +/* + * 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 { eventsModuleKafkaPublishingEventConsumer } from './module'; +import { KafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer'; + +jest.mock('./KafkaPublishingEventConsumer'); + +describe('eventsModuleKafkaPublishingEventConsumer', () => { + it('should be correctly wired and set up', async () => { + const events = new TestEventsService(); + const eventsServiceFactory = createServiceFactory({ + service: eventsServiceRef, + deps: {}, + async factory({}) { + return events; + }, + }); + + const mockKafkaPublishingEventConsumer = { + start: jest.fn(), + shutdown: jest.fn(), + } as unknown as KafkaPublishingEventConsumer; + + jest + .mocked(KafkaPublishingEventConsumer.fromConfig) + .mockReturnValue([mockKafkaPublishingEventConsumer]); + + await startTestBackend({ + features: [ + eventsServiceFactory, + eventsModuleKafkaPublishingEventConsumer, + mockServices.rootConfig.factory({ + data: { + events: { + modules: { + kafka: { + kafkaPublishingEventConsumer: { + dev: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: [ + { + topic: 'fake1', + kafka: { + topic: 'topic-A', + }, + }, + { + topic: 'fake2', + kafka: { + topic: 'topic-B', + }, + }, + ], + }, + }, + }, + }, + }, + }, + }), + ], + }); + + // Verify that the Kafka publishing event consumer was started + expect(mockKafkaPublishingEventConsumer.start).toHaveBeenCalled(); + + // Verify that the shutdown hook was registered (but not called yet) + expect(mockKafkaPublishingEventConsumer.shutdown).not.toHaveBeenCalled(); + }); + + it('should handle empty configuration gracefully', async () => { + const events = new TestEventsService(); + const eventsServiceFactory = createServiceFactory({ + service: eventsServiceRef, + deps: {}, + async factory({}) { + return events; + }, + }); + + jest.mocked(KafkaPublishingEventConsumer.fromConfig).mockReturnValue([]); + + await startTestBackend({ + features: [ + eventsServiceFactory, + eventsModuleKafkaPublishingEventConsumer, + mockServices.rootConfig.factory({ + data: { + events: { + modules: { + kafka: { + // No kafkaPublishingEventConsumer config + }, + }, + }, + }, + }), + ], + }); + + // Verify that fromConfig was called but returned empty array + expect(KafkaPublishingEventConsumer.fromConfig).toHaveBeenCalled(); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.ts b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.ts new file mode 100644 index 0000000000..dd5f0c50ab --- /dev/null +++ b/plugins/events-backend-module-kafka/src/KafkaPublishingEventConsumer/module.ts @@ -0,0 +1,56 @@ +/* + * 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 { eventsServiceRef } from '@backstage/plugin-events-node'; +import { KafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer'; + +/** + * Reads internal Backstage events and forwards them to Kafka topics. + * + * @public + */ +export const eventsModuleKafkaPublishingEventConsumer = createBackendModule({ + pluginId: 'events', + moduleId: 'kafka-publishing-event-consumer', + register(env) { + env.registerInit({ + deps: { + config: coreServices.rootConfig, + events: eventsServiceRef, + logger: coreServices.logger, + lifecycle: coreServices.rootLifecycle, + }, + async init({ config, logger, events, lifecycle }) { + const consumers = KafkaPublishingEventConsumer.fromConfig({ + config, + events, + logger, + }); + + lifecycle.addStartupHook(async () => { + await Promise.all(consumers.map(consumer => consumer.start())); + }); + + lifecycle.addShutdownHook(async () => { + await Promise.all(consumers.map(consumer => consumer.shutdown())); + }); + }, + }); + }, +}); diff --git a/plugins/events-backend-module-kafka/src/index.ts b/plugins/events-backend-module-kafka/src/index.ts index 3ba6207d7d..af4fb02bf1 100644 --- a/plugins/events-backend-module-kafka/src/index.ts +++ b/plugins/events-backend-module-kafka/src/index.ts @@ -13,14 +13,22 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +import { createBackendFeatureLoader } from '@backstage/backend-plugin-api'; +import { eventsModuleKafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher'; +import { eventsModuleKafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer'; /** * 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. + * adding Kafka-based event handling: + * - Consumer: receives events from Kafka topics and passes them to the internal event broker + * - Publisher: receives internal events and publishes them to Kafka topics * * @packageDocumentation */ -export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher'; +export default createBackendFeatureLoader({ + *loader() { + yield eventsModuleKafkaConsumingEventPublisher; + yield eventsModuleKafkaPublishingEventConsumer; + }, +}); diff --git a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts b/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts deleted file mode 100644 index 5824bf4d41..0000000000 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.test.ts +++ /dev/null @@ -1,132 +0,0 @@ -/* - * 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 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, - 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, - }); - - expect(client).toBeDefined(); - - 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, - }); - - 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 deleted file mode 100644 index d506e4a6af..0000000000 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumerClient.ts +++ /dev/null @@ -1,77 +0,0 @@ -/* - * 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 { loggerServiceAdapter } from './LoggerServiceAdapter'; - -/** - * KafkaConsumerClient - * - * This class creates the Kafka client that will be used to create the KafkaConsumingEventPublisher - */ -export class KafkaConsumerClient { - private readonly kafka: Kafka; - private readonly consumers: KafkaConsumingEventPublisher[]; - - static fromConfig(options: { - config: Config; - events: EventsService; - logger: LoggerService; - }): 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( - logger: LoggerService, - events: EventsService, - config: KafkaEventSourceConfig, - ) { - this.kafka = new Kafka({ - ...config.kafkaConfig, - logCreator: loggerServiceAdapter(logger), - }); - - this.consumers = config.kafkaConsumerConfigs.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 deleted file mode 100644 index 64c53595ad..0000000000 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.test.ts +++ /dev/null @@ -1,92 +0,0 @@ -/* - * 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', - }, - consumerSubscribeTopics: { - 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.consumerSubscribeTopics, - ); - 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 deleted file mode 100644 index dfc6186fb5..0000000000 --- a/plugins/events-backend-module-kafka/src/publisher/KafkaConsumingEventPublisher.ts +++ /dev/null @@ -1,109 +0,0 @@ -/* - * 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']; - -/** - * 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. - */ -export class KafkaConsumingEventPublisher { - private readonly kafkaConsumer: Consumer; - private readonly consumerSubscribeTopics: 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 readonly events: EventsService; - - private constructor( - kafkaClient: Kafka, - logger: LoggerService, - events: EventsService, - config: KafkaConsumerConfig, - ) { - this.events = events; - this.kafkaConsumer = kafkaClient.consumer(config.consumerConfig); - 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.consumerSubscribeTopics.topics.toString(), - backstageTopic: config.backstageTopic, - taskId: id, - }); - } - - async start(): Promise { - try { - await this.kafkaConsumer.connect(); - - await this.kafkaConsumer.subscribe(this.consumerSubscribeTopics); - - 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(); - } - - 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/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts deleted file mode 100644 index fc7e3049b0..0000000000 --- a/plugins/events-backend-module-kafka/src/publisher/config.test.ts +++ /dev/null @@ -1,250 +0,0 @@ -/* - * 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 publisherConfigs = readConfig(new ConfigReader({})); - - expect(publisherConfigs).toBeUndefined(); - }); - - 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).toBeDefined(); - - 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( - 'fake1', - ); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.groupId, - ).toEqual('my-group'); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerSubscribeTopics.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: { milliseconds: 20000 }, - initialRetryTime: { milliseconds: 200 }, - factor: '0.4', - multiplier: '4', - retries: '10', - }, - authenticationTimeout: { milliseconds: 20000 }, - connectionTimeout: { milliseconds: 1500 }, - requestTimeout: { milliseconds: 20000 }, - enforceRequestTimeout: false, - topics: [ - { - topic: 'fake1', - kafka: { - topics: ['topic-A'], - groupId: 'my-group', - sessionTimeout: { milliseconds: 20000 }, - rebalanceTimeout: { milliseconds: 50000 }, - heartbeatInterval: { milliseconds: 2000 }, - metadataMaxAge: { milliseconds: 400000 }, - maxBytesPerPartition: 50000, - minBytes: 2, - maxBytes: 500000, - maxWaitTime: { milliseconds: 4000 }, - }, - }, - { - topic: 'fake2', - kafka: { - topics: ['topic-B'], - groupId: 'my-group', - }, - }, - ], - }, - }, - }, - }, - }); - - 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.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?.kafkaConsumerConfigs.length).toBe(2); - expect(publisherConfigs?.kafkaConsumerConfigs[0].backstageTopic).toEqual( - 'fake1', - ); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.groupId, - ).toEqual('my-group'); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerSubscribeTopics.topics, - ).toEqual(['topic-A']); - - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.sessionTimeout, - ).toBe(20000); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.rebalanceTimeout, - ).toBe(50000); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig - .heartbeatInterval, - ).toBe(2000); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.metadataMaxAge, - ).toBe(400000); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig - .maxBytesPerPartition, - ).toBe(50000); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.minBytes, - ).toBe(2); - expect( - publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxBytes, - ).toBe(500000); - expect( - 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 deleted file mode 100644 index 7534ce4d38..0000000000 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ /dev/null @@ -1,153 +0,0 @@ -/* - * 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, readDurationFromConfig } from '@backstage/config'; -import { durationToMilliseconds } from '@backstage/types'; -import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs'; - -export interface KafkaConsumerConfig { - backstageTopic: string; - consumerConfig: ConsumerConfig; - consumerSubscribeTopics: ConsumerSubscribeTopics; -} - -export interface KafkaEventSourceConfig { - kafkaConfig: KafkaConfig; - kafkaConsumerConfigs: KafkaConsumerConfig[]; -} - -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 durationToMilliseconds(humanDuration); -}; - -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'); - - const authenticationTimeout = readOptionalHumanDurationInMs( - kafkaConfig, - 'authenticationTimeout', - ); - - 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['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') - .map(topic => { - return { - backstageTopic: topic.getString('topic'), - consumerConfig: { - groupId: topic.getString('kafka.groupId'), - 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: readOptionalHumanDurationInMs( - topic, - 'kafka.maxWaitTime', - ), - }, - consumerSubscribeTopics: { - topics: topic.getStringArray('kafka.topics'), - }, - }; - }); - - return { - kafkaConfig: { - clientId, - brokers, - ssl, - sasl, - authenticationTimeout, - connectionTimeout, - requestTimeout, - enforceRequestTimeout, - retry, - }, - kafkaConsumerConfigs, - }; -}; diff --git a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts b/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts deleted file mode 100644 index 816793f635..0000000000 --- a/plugins/events-backend-module-kafka/src/service/eventsModuleKafkaConsumingEventPublisher.test.ts +++ /dev/null @@ -1,85 +0,0 @@ -/* - * 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/publisher/LoggerServiceAdapter.ts b/plugins/events-backend-module-kafka/src/utils/LoggerServiceAdapter.ts similarity index 100% rename from plugins/events-backend-module-kafka/src/publisher/LoggerServiceAdapter.ts rename to plugins/events-backend-module-kafka/src/utils/LoggerServiceAdapter.ts diff --git a/plugins/events-backend-module-kafka/src/utils/config.test.ts b/plugins/events-backend-module-kafka/src/utils/config.test.ts new file mode 100644 index 0000000000..2bd6121ba2 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/utils/config.test.ts @@ -0,0 +1,235 @@ +/* + * 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 { readRetryConfig, readKafkaConfig } from './config'; + +describe('readRetryConfig', () => { + it('should return empty object when config is undefined', () => { + const result = readRetryConfig(undefined); + expect(result).toEqual({}); + }); + + it('should return empty object when config is empty', () => { + const config = new ConfigReader({}); + const result = readRetryConfig(config); + expect(result).toEqual({}); + }); + + it('should read retry configuration with milliseconds', () => { + const config = new ConfigReader({ + maxRetryTime: { milliseconds: 20000 }, + initialRetryTime: { milliseconds: 200 }, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + const result = readRetryConfig(config); + + expect(result).toEqual({ + maxRetryTime: 20000, + initialRetryTime: 200, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + }); + + it('should read retry configuration with string values', () => { + const config = new ConfigReader({ + maxRetryTime: { seconds: 20 }, + initialRetryTime: { minutes: 1 }, + factor: '0.4', + multiplier: '4', + retries: '10', + }); + + const result = readRetryConfig(config); + + expect(result).toEqual({ + maxRetryTime: 20000, + initialRetryTime: 60000, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + }); + + it('should handle HumanDuration values', () => { + const config = new ConfigReader({ + maxRetryTime: { hours: 1 }, + initialRetryTime: { days: 1 }, + }); + + const result = readRetryConfig(config); + + expect(result).toEqual({ + maxRetryTime: 3600000, + initialRetryTime: 86400000, + factor: undefined, + multiplier: undefined, + retries: undefined, + }); + }); + + it('should handle partial configuration', () => { + const config = new ConfigReader({ + maxRetryTime: { milliseconds: 15000 }, + retries: 5, + }); + + const result = readRetryConfig(config); + + expect(result).toEqual({ + maxRetryTime: 15000, + initialRetryTime: undefined, + factor: undefined, + multiplier: undefined, + retries: 5, + }); + }); +}); + +describe('readKafkaConfig', () => { + it('should read minimal kafka configuration', () => { + const config = new ConfigReader({ + clientId: 'test-client', + brokers: ['kafka1:9092', 'kafka2:9092'], + }); + + const result = readKafkaConfig(config); + + expect(result).toEqual({ + clientId: 'test-client', + brokers: ['kafka1:9092', 'kafka2:9092'], + authenticationTimeout: undefined, + connectionTimeout: undefined, + requestTimeout: undefined, + enforceRequestTimeout: undefined, + ssl: undefined, + sasl: undefined, + retry: {}, + }); + }); + + it('should read full kafka configuration with all optional fields', () => { + const config = new ConfigReader({ + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + authenticationTimeout: { milliseconds: 20000 }, + connectionTimeout: { milliseconds: 1500 }, + requestTimeout: { milliseconds: 20000 }, + enforceRequestTimeout: false, + ssl: true, + sasl: { + mechanism: 'plain', + username: 'username', + password: 'password', + }, + retry: { + maxRetryTime: { milliseconds: 20000 }, + initialRetryTime: { milliseconds: 200 }, + factor: 0.4, + multiplier: 4, + retries: 10, + }, + }); + + const result = readKafkaConfig(config); + + expect(result).toEqual({ + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + authenticationTimeout: 20000, + connectionTimeout: 1500, + requestTimeout: 20000, + enforceRequestTimeout: false, + ssl: true, + sasl: { + mechanism: 'plain', + username: 'username', + password: 'password', + }, + retry: { + maxRetryTime: 20000, + initialRetryTime: 200, + factor: 0.4, + multiplier: 4, + retries: 10, + }, + }); + }); + + it('should handle HumanDuration values for timeouts', () => { + const config = new ConfigReader({ + clientId: 'test-client', + brokers: ['kafka:9092'], + authenticationTimeout: { hours: 1 }, + connectionTimeout: { days: 1 }, + requestTimeout: { minutes: 5 }, + }); + + const result = readKafkaConfig(config); + + expect(result.authenticationTimeout).toBe(3600000); + expect(result.connectionTimeout).toBe(86400000); + expect(result.requestTimeout).toBe(300000); + }); + + it('should handle complex SSL configuration', () => { + const config = new ConfigReader({ + clientId: 'secure-client', + brokers: ['secure-kafka:9093'], + ssl: { + rejectUnauthorized: false, + ca: 'ca-certificate', + key: 'client-key', + cert: 'client-cert', + }, + }); + + const result = readKafkaConfig(config); + + expect(result.ssl).toEqual({ + rejectUnauthorized: false, + ca: 'ca-certificate', + key: 'client-key', + cert: 'client-cert', + }); + }); + + it('should handle complex SASL configuration', () => { + const config = new ConfigReader({ + clientId: 'sasl-client', + brokers: ['sasl-kafka:9094'], + sasl: { + mechanism: 'scram-sha-256', + username: 'kafka-user', + password: 'kafka-password', + authorizationIdentity: 'authz-user', + }, + }); + + const result = readKafkaConfig(config); + + expect(result.sasl).toEqual({ + mechanism: 'scram-sha-256', + username: 'kafka-user', + password: 'kafka-password', + authorizationIdentity: 'authz-user', + }); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/utils/config.ts b/plugins/events-backend-module-kafka/src/utils/config.ts new file mode 100644 index 0000000000..4e140d26a3 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/utils/config.ts @@ -0,0 +1,84 @@ +/* + * 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, readDurationFromConfig } from '@backstage/config'; +import { durationToMilliseconds } from '@backstage/types'; +import { KafkaConfig, RetryOptions } from 'kafkajs'; + +/** + * 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. + */ +export const readOptionalHumanDurationInMs = ( + config: Config, + key: string, +): number | undefined => { + const humanDuration = config.has(key) + ? readDurationFromConfig(config, { key }) + : undefined; + + if (!humanDuration) return undefined; + + return durationToMilliseconds(humanDuration); +}; + +/** + * Reads retry configuration options from the provided config object. + * + * @param config - The configuration object to read retry options from, or undefined. + * @returns A RetryOptions object with optional retry settings, or an empty object if config is undefined. + */ +export const readRetryConfig = (config: Config | undefined): RetryOptions => { + if (!config) { + return {}; + } + + return { + maxRetryTime: readOptionalHumanDurationInMs(config, 'maxRetryTime'), + initialRetryTime: readOptionalHumanDurationInMs(config, 'initialRetryTime'), + factor: config.getOptionalNumber('factor'), + multiplier: config.getOptionalNumber('multiplier'), + retries: config.getOptionalNumber('retries'), + }; +}; + +/** + * Reads Kafka configuration from the provided config object. + * + * @param config - The configuration object containing Kafka settings. + * @returns A KafkaConfig object with all necessary Kafka connection and authentication settings. + */ +export const readKafkaConfig = (config: Config): KafkaConfig => { + return { + clientId: config.getString('clientId'), + brokers: config.getStringArray('brokers'), + authenticationTimeout: readOptionalHumanDurationInMs( + config, + 'authenticationTimeout', + ), + connectionTimeout: readOptionalHumanDurationInMs( + config, + 'connectionTimeout', + ), + requestTimeout: readOptionalHumanDurationInMs(config, 'requestTimeout'), + enforceRequestTimeout: config.getOptionalBoolean('enforceRequestTimeout'), + ssl: config.getOptional('ssl') as KafkaConfig['ssl'], + sasl: config.getOptional('sasl') as KafkaConfig['sasl'], + retry: readRetryConfig(config.getOptionalConfig('retry')), + }; +}; diff --git a/plugins/events-backend-module-kafka/src/utils/kafkaTransformers.test.ts b/plugins/events-backend-module-kafka/src/utils/kafkaTransformers.test.ts new file mode 100644 index 0000000000..bfb373ea53 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/utils/kafkaTransformers.test.ts @@ -0,0 +1,137 @@ +/* + * 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 { convertHeadersToMetadata, payloadToBuffer } from './kafkaTransformers'; + +describe('kafka-transformers', () => { + describe('convertHeadersToMetadata', () => { + it('should return undefined when headers is undefined', () => { + const result = convertHeadersToMetadata(undefined); + expect(result).toBeUndefined(); + }); + + it('should convert string headers to metadata', () => { + const headers = { + 'content-type': 'application/json', + 'user-id': '12345', + }; + + const result = convertHeadersToMetadata(headers); + + expect(result).toEqual({ + 'content-type': 'application/json', + 'user-id': '12345', + }); + }); + + it('should convert Buffer headers to string metadata', () => { + const headers = { + 'content-type': Buffer.from('application/json'), + 'correlation-id': Buffer.from('abc-123'), + }; + + const result = convertHeadersToMetadata(headers); + + expect(result).toEqual({ + 'content-type': 'application/json', + 'correlation-id': 'abc-123', + }); + }); + + it('should convert array headers to string array metadata', () => { + const headers = { + tags: ['tag1', 'tag2'], + 'buffer-tags': [Buffer.from('tag3'), Buffer.from('tag4')], + 'mixed-tags': ['tag5', Buffer.from('tag6')], + }; + + const result = convertHeadersToMetadata(headers); + + expect(result).toEqual({ + tags: ['tag1', 'tag2'], + 'buffer-tags': ['tag3', 'tag4'], + 'mixed-tags': ['tag5', 'tag6'], + }); + }); + + it('should handle mixed header types', () => { + const headers = { + 'string-header': 'value', + 'buffer-header': Buffer.from('buffer-value'), + 'array-header': ['item1', Buffer.from('item2')], + 'undefined-header': undefined, + }; + + const result = convertHeadersToMetadata(headers); + + expect(result).toEqual({ + 'string-header': 'value', + 'buffer-header': 'buffer-value', + 'array-header': ['item1', 'item2'], + 'undefined-header': undefined, + }); + }); + + it('should handle empty headers object', () => { + const headers = {}; + const result = convertHeadersToMetadata(headers); + expect(result).toEqual({}); + }); + }); + + describe('payloadToBuffer', () => { + it('should return the same Buffer when payload is already a Buffer', () => { + const originalBuffer = Buffer.from('test data'); + + const result = payloadToBuffer(originalBuffer); + + expect(result).toBe(originalBuffer); + expect(Buffer.isBuffer(result)).toBe(true); + }); + + it('should convert string to Buffer', () => { + const payload = 'hello world'; + + const result = payloadToBuffer(payload); + + expect(Buffer.isBuffer(result)).toBe(true); + expect(result.toString()).toBe('hello world'); + }); + + it('should convert object to JSON Buffer', () => { + const payload = { name: 'John', age: 30 }; + + const result = payloadToBuffer(payload); + + expect(Buffer.isBuffer(result)).toBe(true); + expect(JSON.parse(result.toString())).toEqual(payload); + }); + + it('should convert array to JSON Buffer', () => { + const payload = [1, 2, 3, 'test']; + + const result = payloadToBuffer(payload); + + expect(Buffer.isBuffer(result)).toBe(true); + expect(JSON.parse(result.toString())).toEqual(payload); + }); + + it('should convert primitives to JSON Buffer', () => { + expect(payloadToBuffer(42).toString()).toBe('42'); + expect(payloadToBuffer(true).toString()).toBe('true'); + expect(payloadToBuffer(null).toString()).toBe('null'); + }); + }); +}); diff --git a/plugins/events-backend-module-kafka/src/utils/kafkaTransformers.ts b/plugins/events-backend-module-kafka/src/utils/kafkaTransformers.ts new file mode 100644 index 0000000000..871cb08a04 --- /dev/null +++ b/plugins/events-backend-module-kafka/src/utils/kafkaTransformers.ts @@ -0,0 +1,49 @@ +/* + * 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 { EventParams } from '@backstage/plugin-events-node'; +import { IHeaders } from 'kafkajs'; + +type EventMetadata = EventParams['metadata']; + +export const convertHeadersToMetadata = ( + headers: IHeaders | undefined, +): EventMetadata => { + 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; +}; + +export const payloadToBuffer = (payload: unknown): Buffer => { + if (Buffer.isBuffer(payload)) { + return payload; + } + + if (typeof payload === 'string') { + return Buffer.from(payload, 'utf8'); // More explicit encoding + } + + // Convert to JSON string then encode + return Buffer.from(JSON.stringify(payload), 'utf8'); +}; From 88a1dce3c5806ccd053b7ce1a78c4b4005d3ec04 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Mon, 8 Dec 2025 13:08:00 +0100 Subject: [PATCH 2/5] docs(events): update readme documentation Signed-off-by: Jonas Beck --- plugins/events-backend-module-kafka/README.md | 92 ++++++++++++++----- 1 file changed, 67 insertions(+), 25 deletions(-) diff --git a/plugins/events-backend-module-kafka/README.md b/plugins/events-backend-module-kafka/README.md index dbc5bf5607..f77b5d2f21 100644 --- a/plugins/events-backend-module-kafka/README.md +++ b/plugins/events-backend-module-kafka/README.md @@ -2,65 +2,107 @@ 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 package is a module for the `events-backend` backend plugin and extends the events system with a `KafkaConsumingEventPublisher` and `KafkaPublishingEventConsumer` -This event publisher will allow you to receive events from an Kafka queue and will publish these to the used `EventsService` implementation. +This module provides two-way integration with Kafka: + +- **KafkaConsumingEventPublisher**: Receives events from Kafka queues and publishes them to the Backstage events system +- **KafkaPublishingEventConsumer**: Consumes events from Backstage and publishes them to Kafka queues ## Configuration -To set up Kafka queues, you need to configure the following values: +To set up Kafka integration, you need to configure one or both of the following components: + +### KafkaConsumingEventPublisher Configuration + +To receive events from Kafka queues and publish them to Backstage: ```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. + production: # Instance name, will be included in logs + 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. +``` + +### KafkaPublishingEventConsumer Configuration + +To publish events from Backstage to Kafka queues, you can configure the `KafkaPublishingEventConsumer`: + +```yaml +events: + modules: + kafka: + kafkaPublishingEventConsumer: + production: # Instance name, will be included in logs + 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: 'catalog.entity.created' # (Required) The Backstage topic to consume from + kafka: + topic: kafka-topic-name # (Required) The Kafka topic to publish to ``` 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 -If your Kafka cluster requires SSL, you can configure it as follows: +If your Kafka cluster requires SSL, you can configure it for both `kafkaConsumingEventPublisher` and `kafkaPublishingEventConsumer` instances: ```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. + production: + # ... other configuration ... + 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. + kafkaPublishingEventConsumer: + production: + # ... other configuration ... + ssl: + # Same SSL configuration options as above ``` ### Optional SASL Authentication Configuration -If your Kafka cluster requires `SASL` authentication, you can configure it as follows: +If your Kafka cluster requires SASL authentication, you can configure it for both components: ```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 + production: + # ... other configuration ... + sasl: + mechanism: 'plain' # SASL mechanism ('plain', 'scram-sha-256' or 'scram-sha-512') + username: your-username # SASL username + password: your-password # SASL password + kafkaPublishingEventConsumer: + production: + # ... other configuration ... + sasl: + # Same SASL configuration options as above ``` -This section includes optional `SSL` and `SASL` authentication configuration for enhanced security. +These SSL and SASL configurations apply to both Kafka components and provide enhanced security for your Kafka connections. ## Installation From 2c74ea97b5cd13862b195eb301c299422c383873 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Mon, 8 Dec 2025 13:17:48 +0100 Subject: [PATCH 3/5] chore(changeset): add changesets Signed-off-by: Jonas Beck --- .changeset/angry-views-win.md | 23 +++++++++++++++++++++++ .changeset/stale-poets-battle.md | 7 +++++++ 2 files changed, 30 insertions(+) create mode 100644 .changeset/angry-views-win.md create mode 100644 .changeset/stale-poets-battle.md diff --git a/.changeset/angry-views-win.md b/.changeset/angry-views-win.md new file mode 100644 index 0000000000..bf45881d25 --- /dev/null +++ b/.changeset/angry-views-win.md @@ -0,0 +1,23 @@ +--- +'@backstage/plugin-events-backend-module-kafka': minor +--- + +**BREAKING**: Updated `kafkaConsumingEventPublisher` configuration to support multiple named instances + +The Kafka configuration now requires named instances instead of a single configuration object for `kafkaConsumingEventPublisher`, this allows for multiple Kafka configurations. + +These changes are **required** to your `app-config.yaml`: + +```diff +events: + modules: + kafka: + kafkaConsumingEventPublisher: +- clientId: your-client-id +- brokers: [...] +- topics: [...] ++ default: # Or any name like 'prod', 'dev', etc. ++ clientId: your-client-id ++ brokers: [...] ++ topics: [...] +``` diff --git a/.changeset/stale-poets-battle.md b/.changeset/stale-poets-battle.md new file mode 100644 index 0000000000..1725fd2885 --- /dev/null +++ b/.changeset/stale-poets-battle.md @@ -0,0 +1,7 @@ +--- +'@backstage/plugin-events-backend-module-kafka': minor +--- + +Added `KafkaPublishingEventConsumer` to support sending Backstage events to Kafka topics. + +This addition enables Backstage to publish events to external Kafka systems, complementing the existing ability to receive events from Kafka. This allows for better integration with external systems that rely on Kafka for event streaming. From d3d33a08706d499716dbd9b53d3d22d8df9d3ccf Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Tue, 9 Dec 2025 10:04:16 +0100 Subject: [PATCH 4/5] chore: update api report Signed-off-by: Jonas Beck --- plugins/events-backend-module-kafka/report.api.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/events-backend-module-kafka/report.api.md b/plugins/events-backend-module-kafka/report.api.md index a6b129b82e..c5e6d29c6a 100644 --- a/plugins/events-backend-module-kafka/report.api.md +++ b/plugins/events-backend-module-kafka/report.api.md @@ -6,6 +6,6 @@ import { BackendFeature } from '@backstage/backend-plugin-api'; // @public -const eventsModuleKafkaConsumingEventPublisher: BackendFeature; -export default eventsModuleKafkaConsumingEventPublisher; +const _default: BackendFeature; +export default _default; ``` From 9fa23b3ea1f5724800cf3acb243137dc13e85eb4 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Thu, 11 Dec 2025 10:16:46 +0100 Subject: [PATCH 5/5] feat(events): add backwards compatability to kafkaConsumingEventPublisher config Signed-off-by: Jonas Beck --- .changeset/angry-views-win.md | 20 +- .../events-backend-module-kafka/config.d.ts | 487 ++++++++++++------ .../KafkaConsumingEventPublisher.ts | 2 +- .../config.test.ts | 87 +++- .../KafkaConsumingEventPublisher/config.ts | 119 +++-- 5 files changed, 488 insertions(+), 227 deletions(-) diff --git a/.changeset/angry-views-win.md b/.changeset/angry-views-win.md index bf45881d25..9d406280a7 100644 --- a/.changeset/angry-views-win.md +++ b/.changeset/angry-views-win.md @@ -2,22 +2,4 @@ '@backstage/plugin-events-backend-module-kafka': minor --- -**BREAKING**: Updated `kafkaConsumingEventPublisher` configuration to support multiple named instances - -The Kafka configuration now requires named instances instead of a single configuration object for `kafkaConsumingEventPublisher`, this allows for multiple Kafka configurations. - -These changes are **required** to your `app-config.yaml`: - -```diff -events: - modules: - kafka: - kafkaConsumingEventPublisher: -- clientId: your-client-id -- brokers: [...] -- topics: [...] -+ default: # Or any name like 'prod', 'dev', etc. -+ clientId: your-client-id -+ brokers: [...] -+ topics: [...] -``` +Added support for multiple named instances in `kafkaConsumingEventPublisher` configuration. The previous single configuration format is still supported for backward compatibility. diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts index b3698a8999..47ebae0653 100644 --- a/plugins/events-backend-module-kafka/config.d.ts +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -25,184 +25,359 @@ export interface Config { /** * Configuration for KafkaConsumingEventPublisher * - * Supports multiple named instances as a record where each key is a unique name - * for the Kafka consumer configuration. + * Supports either: + * 1. Single configuration object (legacy format) + * 2. Multiple named instances as a record where each key is a unique name for the Kafka instance */ - kafkaConsumingEventPublisher?: { - [name: string]: { - /** - * (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?: { + kafkaConsumingEventPublisher?: + | { /** - * (Optional) Maximum wait time for a retry - * Default: 30000 ms. + * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. */ - maxRetryTime?: HumanDuration | string; + 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) Initial value used to calculate the retry (This is still randomized following the randomization factor) - * Default: 300 ms. + * Optional retry connection parameters. */ - initialRetryTime?: HumanDuration | string; - - /** - * (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 for authentication requests. - * Default: 10000 ms. - */ - authenticationTimeout?: HumanDuration | string; - - /** - * (Optional) Time to wait for a successful connection. - * Default: 1000 ms. - */ - connectionTimeout?: HumanDuration | string; - - /** - * (Optional) Time to wait for a successful request. - * Default: 30000 ms. - */ - requestTimeout?: HumanDuration | string; - - /** - * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. - * Default: true - */ - enforceRequestTimeout?: boolean; - - /** - * Contains an object per topic for which a 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: { + retry?: { /** - * (Required) The Kafka topics to subscribe to - */ - topics: string[]; - /** - * (Required) The GroupId to be used by the topic consumers - */ - groupId: string; - - /** - * (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 + * (Optional) Maximum wait time for a retry * Default: 30000 ms. */ - sessionTimeout?: HumanDuration | string; + maxRetryTime?: HumanDuration | string; /** - * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group - * Default: 60000 ms. + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. */ - rebalanceTimeout?: HumanDuration | string; + initialRetryTime?: HumanDuration | string; /** - * (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. + * (Optional) Randomization factor + * Default: 0.2. */ - heartbeatInterval?: HumanDuration | string; + factor?: number; /** - * (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). + * (Optional) Exponential factor + * Default: 2. */ - metadataMaxAge?: HumanDuration | string; + multiplier?: 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) + * (Optional) Max number of retries per call + * Default: 5. */ - maxBytesPerPartition?: number; - - /** - * (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; - - /** - * (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 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 - */ - maxWaitTime?: HumanDuration | string; + retries?: number; }; - }>; - }; - }; + + /** + * (Optional) Timeout for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout?: HumanDuration | string; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout?: boolean; + + /** + * Contains an object per topic for which a 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 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?: HumanDuration | string; + + /** + * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group + * Default: 60000 ms. + */ + rebalanceTimeout?: HumanDuration | string; + + /** + * (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?: HumanDuration | string; + + /** + * (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?: HumanDuration | string; + + /** + * (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 maxWaitTime 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 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 + */ + maxWaitTime?: HumanDuration | string; + }; + }>; + } + | { + [name: string]: { + /** + * (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 + * Default: 30000 ms. + */ + maxRetryTime?: HumanDuration | string; + + /** + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. + */ + initialRetryTime?: HumanDuration | string; + + /** + * (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 for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout?: HumanDuration | string; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout?: boolean; + + /** + * Contains an object per topic for which a 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 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?: HumanDuration | string; + + /** + * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group + * Default: 60000 ms. + */ + rebalanceTimeout?: HumanDuration | string; + + /** + * (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?: HumanDuration | string; + + /** + * (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?: HumanDuration | string; + + /** + * (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 maxWaitTime 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 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 + */ + maxWaitTime?: HumanDuration | string; + }; + }>; + }; + }; /** * Configuration for KafkaPublishingEventConsumer diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts index 41744abff2..b996032575 100644 --- a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts @@ -43,7 +43,7 @@ export class KafkaConsumingEventPublisher { events: EventsService; logger: LoggerService; }): KafkaConsumingEventPublisher[] { - const configs = readConsumerConfig(env.config); + const configs = readConsumerConfig(env.config, env.logger); return configs.map( kafkaConfig => diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts index b61ee9332e..16ee9e0a4c 100644 --- a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts @@ -15,10 +15,16 @@ */ import { ConfigReader } from '@backstage/config'; import { readConsumerConfig } from './config'; +import { mockServices } from '@backstage/backend-test-utils'; + +const mockLogger = mockServices.logger.mock(); describe('readConsumerConfig', () => { it('not configured', () => { - const publisherConfigs = readConsumerConfig(new ConfigReader({})); + const publisherConfigs = readConsumerConfig( + new ConfigReader({}), + mockLogger, + ); expect(publisherConfigs).toEqual([]); }); @@ -55,7 +61,7 @@ describe('readConsumerConfig', () => { }, }); - const publisherConfigs = readConsumerConfig(config); + const publisherConfigs = readConsumerConfig(config, mockLogger); expect(publisherConfigs).toBeDefined(); expect(Array.isArray(publisherConfigs)).toBe(true); @@ -150,7 +156,7 @@ describe('readConsumerConfig', () => { }, }); - const publisherConfigs = readConsumerConfig(config); + const publisherConfigs = readConsumerConfig(config, mockLogger); expect(publisherConfigs).toBeDefined(); expect(Array.isArray(publisherConfigs)).toBe(true); @@ -243,7 +249,7 @@ describe('readConsumerConfig', () => { }, }); - const publisherConfigs = readConsumerConfig(config); + const publisherConfigs = readConsumerConfig(config, mockLogger); expect(publisherConfigs).toBeDefined(); expect(Array.isArray(publisherConfigs)).toBe(true); @@ -272,4 +278,77 @@ describe('readConsumerConfig', () => { // Consumer configuration expect(devConfig.kafkaConsumerConfigs.length).toBe(0); }); + + it('single instance configuration (legacy format)', () => { + 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 = readConsumerConfig(config, mockLogger); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const defaultConfig = publisherConfigs[0]; + expect(defaultConfig.instance).toBe('default'); + expect(defaultConfig.kafkaConsumerConfigs.length).toBe(2); + + expect(defaultConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(defaultConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + + expect(defaultConfig.kafkaConsumerConfigs).toEqual([ + { + backstageTopic: 'fake1', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-A'], + }, + }, + { + backstageTopic: 'fake2', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-B'], + }, + }, + ]); + + // Verify deprecation warning was logged + expect(mockLogger.warn).toHaveBeenCalledWith( + 'Legacy single config format detected at events.modules.kafka.kafkaConsumingEventPublisher.', + ); + }); }); diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts index dc9b477770..b07378a1d6 100644 --- a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts @@ -19,6 +19,7 @@ import { readKafkaConfig, readOptionalHumanDurationInMs, } from '../utils/config'; +import { LoggerService } from '@backstage/backend-plugin-api'; export interface KafkaConsumerConfig { backstageTopic: string; @@ -35,57 +36,81 @@ export interface KafkaConsumingEventPublisherConfig { const CONFIG_PREFIX_PUBLISHER = 'events.modules.kafka.kafkaConsumingEventPublisher'; +const processSinglePublisher = ( + instanceName: string, + publisherConfig: Config, +): KafkaConsumingEventPublisherConfig => { + return { + instance: instanceName, + kafkaConfig: readKafkaConfig(publisherConfig), + kafkaConsumerConfigs: publisherConfig + .getConfigArray('topics') + .map(topicConfig => { + return { + backstageTopic: topicConfig.getString('topic'), + consumerConfig: { + groupId: topicConfig.getString('kafka.groupId'), + sessionTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.sessionTimeout', + ), + rebalanceTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.rebalanceTimeout', + ), + heartbeatInterval: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.heartbeatInterval', + ), + metadataMaxAge: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.metadataMaxAge', + ), + maxBytesPerPartition: topicConfig.getOptionalNumber( + 'kafka.maxBytesPerPartition', + ), + minBytes: topicConfig.getOptionalNumber('kafka.minBytes'), + maxBytes: topicConfig.getOptionalNumber('kafka.maxBytes'), + maxWaitTimeInMs: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.maxWaitTime', + ), + }, + consumerSubscribeTopics: { + topics: topicConfig.getStringArray('kafka.topics'), + }, + }; + }), + }; +}; + export const readConsumerConfig = ( config: Config, + logger: LoggerService, ): KafkaConsumingEventPublisherConfig[] => { - const publishers = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + const publishersConfig = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + + // Check for legacy single publisher format + if (publishersConfig?.getOptionalString('clientId')) { + logger.warn( + 'Legacy single config format detected at events.modules.kafka.kafkaConsumingEventPublisher.', + ); + return [ + processSinglePublisher( + 'default', // use `default` as instance name for legacy single config + publishersConfig, + ), + ]; + } return ( - publishers?.keys()?.map(publisherKey => { - const publisherConfig = publishers.getConfig(publisherKey); - - return { - instance: publisherKey, - kafkaConfig: readKafkaConfig(publisherConfig), - kafkaConsumerConfigs: publisherConfig - .getConfigArray('topics') - .map(topicConfig => { - return { - backstageTopic: topicConfig.getString('topic'), - consumerConfig: { - groupId: topicConfig.getString('kafka.groupId'), - sessionTimeout: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.sessionTimeout', - ), - rebalanceTimeout: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.rebalanceTimeout', - ), - heartbeatInterval: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.heartbeatInterval', - ), - metadataMaxAge: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.metadataMaxAge', - ), - maxBytesPerPartition: topicConfig.getOptionalNumber( - 'kafka.maxBytesPerPartition', - ), - minBytes: topicConfig.getOptionalNumber('kafka.minBytes'), - maxBytes: topicConfig.getOptionalNumber('kafka.maxBytes'), - maxWaitTimeInMs: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.maxWaitTime', - ), - }, - consumerSubscribeTopics: { - topics: topicConfig.getStringArray('kafka.topics'), - }, - }; - }), - }; - }) ?? [] + publishersConfig + ?.keys() + ?.map(publisherKey => + processSinglePublisher( + publisherKey, + publishersConfig.getConfig(publisherKey), + ), + ) ?? [] ); };