From 805c48d23d742e0a6bb735f4c6b316b8eb8a3dc7 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Mon, 8 Dec 2025 12:37:02 +0100 Subject: [PATCH] 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'); +};