diff --git a/.changeset/angry-views-win.md b/.changeset/angry-views-win.md index bf45881d25..9d406280a7 100644 --- a/.changeset/angry-views-win.md +++ b/.changeset/angry-views-win.md @@ -2,22 +2,4 @@ '@backstage/plugin-events-backend-module-kafka': minor --- -**BREAKING**: Updated `kafkaConsumingEventPublisher` configuration to support multiple named instances - -The Kafka configuration now requires named instances instead of a single configuration object for `kafkaConsumingEventPublisher`, this allows for multiple Kafka configurations. - -These changes are **required** to your `app-config.yaml`: - -```diff -events: - modules: - kafka: - kafkaConsumingEventPublisher: -- clientId: your-client-id -- brokers: [...] -- topics: [...] -+ default: # Or any name like 'prod', 'dev', etc. -+ clientId: your-client-id -+ brokers: [...] -+ topics: [...] -``` +Added support for multiple named instances in `kafkaConsumingEventPublisher` configuration. The previous single configuration format is still supported for backward compatibility. diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts index b3698a8999..47ebae0653 100644 --- a/plugins/events-backend-module-kafka/config.d.ts +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -25,184 +25,359 @@ export interface Config { /** * Configuration for KafkaConsumingEventPublisher * - * Supports multiple named instances as a record where each key is a unique name - * for the Kafka consumer configuration. + * Supports either: + * 1. Single configuration object (legacy format) + * 2. Multiple named instances as a record where each key is a unique name for the Kafka instance */ - kafkaConsumingEventPublisher?: { - [name: string]: { - /** - * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. - */ - clientId: string; - /** - * (Required) List of brokers in the Kafka cluster to connect to. - */ - brokers: string[]; - /** - * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect. - * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options - */ - ssl?: - | { - ca?: string[]; - /** @visibility secret */ - key?: string; - cert?: string; - rejectUnauthorized?: boolean; - } - | boolean; - /** - * Optional SASL connection parameters. - */ - sasl?: { - mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512'; - username: string; - /** @visibility secret */ - password: string; - }; - - /** - * Optional retry connection parameters. - */ - retry?: { + kafkaConsumingEventPublisher?: + | { /** - * (Optional) Maximum wait time for a retry - * Default: 30000 ms. + * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. */ - maxRetryTime?: HumanDuration | string; + clientId: string; + /** + * (Required) List of brokers in the Kafka cluster to connect to. + */ + brokers: string[]; + /** + * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect. + * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options + */ + ssl?: + | { + ca?: string[]; + /** @visibility secret */ + key?: string; + cert?: string; + rejectUnauthorized?: boolean; + } + | boolean; + /** + * Optional SASL connection parameters. + */ + sasl?: { + mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512'; + username: string; + /** @visibility secret */ + password: string; + }; /** - * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) - * Default: 300 ms. + * Optional retry connection parameters. */ - initialRetryTime?: HumanDuration | string; - - /** - * (Optional) Randomization factor - * Default: 0.2. - */ - factor?: number; - - /** - * (Optional) Exponential factor - * Default: 2. - */ - multiplier?: number; - - /** - * (Optional) Max number of retries per call - * Default: 5. - */ - retries?: number; - }; - - /** - * (Optional) Timeout for authentication requests. - * Default: 10000 ms. - */ - authenticationTimeout?: HumanDuration | string; - - /** - * (Optional) Time to wait for a successful connection. - * Default: 1000 ms. - */ - connectionTimeout?: HumanDuration | string; - - /** - * (Optional) Time to wait for a successful request. - * Default: 30000 ms. - */ - requestTimeout?: HumanDuration | string; - - /** - * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. - * Default: true - */ - enforceRequestTimeout?: boolean; - - /** - * Contains an object per topic for which a Kafka queue - * should be used as source of events. - */ - topics: Array<{ - /** - * (Required) The Backstage topic to publish to - */ - topic: string; - /** - * (Required) KafkaConsumer-related configuration. - */ - kafka: { + retry?: { /** - * (Required) The Kafka topics to subscribe to - */ - topics: string[]; - /** - * (Required) The GroupId to be used by the topic consumers - */ - groupId: string; - - /** - * (Optional) Timeout used to detect failures. - * The consumer sends periodic heartbeats to indicate its liveness to the broker. - * If no heartbeats are received by the broker before the expiration of this session timeout, - * then the broker will remove this consumer from the group and initiate a rebalance + * (Optional) Maximum wait time for a retry * Default: 30000 ms. */ - sessionTimeout?: HumanDuration | string; + maxRetryTime?: HumanDuration | string; /** - * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group - * Default: 60000 ms. + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. */ - rebalanceTimeout?: HumanDuration | string; + initialRetryTime?: HumanDuration | string; /** - * (Optional) The expected time between heartbeats to the consumer coordinator. - * Heartbeats are used to ensure that the consumer's session stays active. - * The value must be set lower than session timeout - * Default: 3000 ms. + * (Optional) Randomization factor + * Default: 0.2. */ - heartbeatInterval?: HumanDuration | string; + factor?: number; /** - * (Optional) The period of time after which we force a refresh of metadata - * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions - * Default: 300000 ms (5 minutes). + * (Optional) Exponential factor + * Default: 2. */ - metadataMaxAge?: HumanDuration | string; + multiplier?: number; /** - * (Optional) The maximum amount of data per-partition the server will return. - * This size must be at least as large as the maximum message size the server allows - * or else it is possible for the producer to send messages larger than the consumer can fetch. - * If that happens, the consumer can get stuck trying to fetch a large message on a certain partition - * Default: 1048576 (1MB) + * (Optional) Max number of retries per call + * Default: 5. */ - maxBytesPerPartition?: number; - - /** - * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate. - * Default: 1 - */ - minBytes?: number; - - /** - * (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0 - * Default: 10485760 (10MB) - */ - maxBytes?: number; - - /** - * (Optional) The maximum amount of time the server will block before answering the fetch request - * if there isn't sufficient data to immediately satisfy the requirement given by minBytes - * Default: 5000 - */ - maxWaitTime?: HumanDuration | string; + retries?: number; }; - }>; - }; - }; + + /** + * (Optional) Timeout for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout?: HumanDuration | string; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout?: boolean; + + /** + * Contains an object per topic for which a Kafka queue + * should be used as source of events. + */ + topics: Array<{ + /** + * (Required) The Backstage topic to publish to + */ + topic: string; + /** + * (Required) KafkaConsumer-related configuration. + */ + kafka: { + /** + * (Required) The Kafka topics to subscribe to + */ + topics: string[]; + /** + * (Required) The GroupId to be used by the topic consumers + */ + groupId: string; + + /** + * (Optional) Timeout used to detect failures. + * The consumer sends periodic heartbeats to indicate its liveness to the broker. + * If no heartbeats are received by the broker before the expiration of this session timeout, + * then the broker will remove this consumer from the group and initiate a rebalance + * Default: 30000 ms. + */ + sessionTimeout?: HumanDuration | string; + + /** + * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group + * Default: 60000 ms. + */ + rebalanceTimeout?: HumanDuration | string; + + /** + * (Optional) The expected time between heartbeats to the consumer coordinator. + * Heartbeats are used to ensure that the consumer's session stays active. + * The value must be set lower than session timeout + * Default: 3000 ms. + */ + heartbeatInterval?: HumanDuration | string; + + /** + * (Optional) The period of time after which we force a refresh of metadata + * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions + * Default: 300000 ms (5 minutes). + */ + metadataMaxAge?: HumanDuration | string; + + /** + * (Optional) The maximum amount of data per-partition the server will return. + * This size must be at least as large as the maximum message size the server allows + * or else it is possible for the producer to send messages larger than the consumer can fetch. + * If that happens, the consumer can get stuck trying to fetch a large message on a certain partition + * Default: 1048576 (1MB) + */ + maxBytesPerPartition?: number; + + /** + * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate. + * Default: 1 + */ + minBytes?: number; + + /** + * (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0 + * Default: 10485760 (10MB) + */ + maxBytes?: number; + + /** + * (Optional) The maximum amount of time the server will block before answering the fetch request + * if there isn't sufficient data to immediately satisfy the requirement given by minBytes + * Default: 5000 + */ + maxWaitTime?: HumanDuration | string; + }; + }>; + } + | { + [name: string]: { + /** + * (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster. + */ + clientId: string; + /** + * (Required) List of brokers in the Kafka cluster to connect to. + */ + brokers: string[]; + /** + * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect. + * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options + */ + ssl?: + | { + ca?: string[]; + /** @visibility secret */ + key?: string; + cert?: string; + rejectUnauthorized?: boolean; + } + | boolean; + /** + * Optional SASL connection parameters. + */ + sasl?: { + mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512'; + username: string; + /** @visibility secret */ + password: string; + }; + + /** + * Optional retry connection parameters. + */ + retry?: { + /** + * (Optional) Maximum wait time for a retry + * Default: 30000 ms. + */ + maxRetryTime?: HumanDuration | string; + + /** + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) + * Default: 300 ms. + */ + initialRetryTime?: HumanDuration | string; + + /** + * (Optional) Randomization factor + * Default: 0.2. + */ + factor?: number; + + /** + * (Optional) Exponential factor + * Default: 2. + */ + multiplier?: number; + + /** + * (Optional) Max number of retries per call + * Default: 5. + */ + retries?: number; + }; + + /** + * (Optional) Timeout for authentication requests. + * Default: 10000 ms. + */ + authenticationTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful connection. + * Default: 1000 ms. + */ + connectionTimeout?: HumanDuration | string; + + /** + * (Optional) Time to wait for a successful request. + * Default: 30000 ms. + */ + requestTimeout?: HumanDuration | string; + + /** + * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. + * Default: true + */ + enforceRequestTimeout?: boolean; + + /** + * Contains an object per topic for which a Kafka queue + * should be used as source of events. + */ + topics: Array<{ + /** + * (Required) The Backstage topic to publish to + */ + topic: string; + /** + * (Required) KafkaConsumer-related configuration. + */ + kafka: { + /** + * (Required) The Kafka topics to subscribe to + */ + topics: string[]; + /** + * (Required) The GroupId to be used by the topic consumers + */ + groupId: string; + + /** + * (Optional) Timeout used to detect failures. + * The consumer sends periodic heartbeats to indicate its liveness to the broker. + * If no heartbeats are received by the broker before the expiration of this session timeout, + * then the broker will remove this consumer from the group and initiate a rebalance + * Default: 30000 ms. + */ + sessionTimeout?: HumanDuration | string; + + /** + * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group + * Default: 60000 ms. + */ + rebalanceTimeout?: HumanDuration | string; + + /** + * (Optional) The expected time between heartbeats to the consumer coordinator. + * Heartbeats are used to ensure that the consumer's session stays active. + * The value must be set lower than session timeout + * Default: 3000 ms. + */ + heartbeatInterval?: HumanDuration | string; + + /** + * (Optional) The period of time after which we force a refresh of metadata + * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions + * Default: 300000 ms (5 minutes). + */ + metadataMaxAge?: HumanDuration | string; + + /** + * (Optional) The maximum amount of data per-partition the server will return. + * This size must be at least as large as the maximum message size the server allows + * or else it is possible for the producer to send messages larger than the consumer can fetch. + * If that happens, the consumer can get stuck trying to fetch a large message on a certain partition + * Default: 1048576 (1MB) + */ + maxBytesPerPartition?: number; + + /** + * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate. + * Default: 1 + */ + minBytes?: number; + + /** + * (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0 + * Default: 10485760 (10MB) + */ + maxBytes?: number; + + /** + * (Optional) The maximum amount of time the server will block before answering the fetch request + * if there isn't sufficient data to immediately satisfy the requirement given by minBytes + * Default: 5000 + */ + maxWaitTime?: HumanDuration | string; + }; + }>; + }; + }; /** * Configuration for KafkaPublishingEventConsumer diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts index 41744abff2..b996032575 100644 --- a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts @@ -43,7 +43,7 @@ export class KafkaConsumingEventPublisher { events: EventsService; logger: LoggerService; }): KafkaConsumingEventPublisher[] { - const configs = readConsumerConfig(env.config); + const configs = readConsumerConfig(env.config, env.logger); return configs.map( kafkaConfig => diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts index b61ee9332e..16ee9e0a4c 100644 --- a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts @@ -15,10 +15,16 @@ */ import { ConfigReader } from '@backstage/config'; import { readConsumerConfig } from './config'; +import { mockServices } from '@backstage/backend-test-utils'; + +const mockLogger = mockServices.logger.mock(); describe('readConsumerConfig', () => { it('not configured', () => { - const publisherConfigs = readConsumerConfig(new ConfigReader({})); + const publisherConfigs = readConsumerConfig( + new ConfigReader({}), + mockLogger, + ); expect(publisherConfigs).toEqual([]); }); @@ -55,7 +61,7 @@ describe('readConsumerConfig', () => { }, }); - const publisherConfigs = readConsumerConfig(config); + const publisherConfigs = readConsumerConfig(config, mockLogger); expect(publisherConfigs).toBeDefined(); expect(Array.isArray(publisherConfigs)).toBe(true); @@ -150,7 +156,7 @@ describe('readConsumerConfig', () => { }, }); - const publisherConfigs = readConsumerConfig(config); + const publisherConfigs = readConsumerConfig(config, mockLogger); expect(publisherConfigs).toBeDefined(); expect(Array.isArray(publisherConfigs)).toBe(true); @@ -243,7 +249,7 @@ describe('readConsumerConfig', () => { }, }); - const publisherConfigs = readConsumerConfig(config); + const publisherConfigs = readConsumerConfig(config, mockLogger); expect(publisherConfigs).toBeDefined(); expect(Array.isArray(publisherConfigs)).toBe(true); @@ -272,4 +278,77 @@ describe('readConsumerConfig', () => { // Consumer configuration expect(devConfig.kafkaConsumerConfigs.length).toBe(0); }); + + it('single instance configuration (legacy format)', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + topics: [ + { + topic: 'fake1', + kafka: { + topics: ['topic-A'], + groupId: 'my-group', + }, + }, + { + topic: 'fake2', + kafka: { + topics: ['topic-B'], + groupId: 'my-group', + }, + }, + ], + }, + }, + }, + }, + }); + + const publisherConfigs = readConsumerConfig(config, mockLogger); + + expect(publisherConfigs).toBeDefined(); + expect(Array.isArray(publisherConfigs)).toBe(true); + expect(publisherConfigs).toHaveLength(1); + + const defaultConfig = publisherConfigs[0]; + expect(defaultConfig.instance).toBe('default'); + expect(defaultConfig.kafkaConsumerConfigs.length).toBe(2); + + expect(defaultConfig.kafkaConfig.clientId).toEqual('backstage-events'); + expect(defaultConfig.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + + expect(defaultConfig.kafkaConsumerConfigs).toEqual([ + { + backstageTopic: 'fake1', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-A'], + }, + }, + { + backstageTopic: 'fake2', + consumerConfig: { + groupId: 'my-group', + }, + consumerSubscribeTopics: { + topics: ['topic-B'], + }, + }, + ]); + + // Verify deprecation warning was logged + expect(mockLogger.warn).toHaveBeenCalledWith( + 'Legacy single config format detected at events.modules.kafka.kafkaConsumingEventPublisher.', + ); + }); }); diff --git a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts index dc9b477770..b07378a1d6 100644 --- a/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts +++ b/plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts @@ -19,6 +19,7 @@ import { readKafkaConfig, readOptionalHumanDurationInMs, } from '../utils/config'; +import { LoggerService } from '@backstage/backend-plugin-api'; export interface KafkaConsumerConfig { backstageTopic: string; @@ -35,57 +36,81 @@ export interface KafkaConsumingEventPublisherConfig { const CONFIG_PREFIX_PUBLISHER = 'events.modules.kafka.kafkaConsumingEventPublisher'; +const processSinglePublisher = ( + instanceName: string, + publisherConfig: Config, +): KafkaConsumingEventPublisherConfig => { + return { + instance: instanceName, + kafkaConfig: readKafkaConfig(publisherConfig), + kafkaConsumerConfigs: publisherConfig + .getConfigArray('topics') + .map(topicConfig => { + return { + backstageTopic: topicConfig.getString('topic'), + consumerConfig: { + groupId: topicConfig.getString('kafka.groupId'), + sessionTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.sessionTimeout', + ), + rebalanceTimeout: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.rebalanceTimeout', + ), + heartbeatInterval: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.heartbeatInterval', + ), + metadataMaxAge: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.metadataMaxAge', + ), + maxBytesPerPartition: topicConfig.getOptionalNumber( + 'kafka.maxBytesPerPartition', + ), + minBytes: topicConfig.getOptionalNumber('kafka.minBytes'), + maxBytes: topicConfig.getOptionalNumber('kafka.maxBytes'), + maxWaitTimeInMs: readOptionalHumanDurationInMs( + topicConfig, + 'kafka.maxWaitTime', + ), + }, + consumerSubscribeTopics: { + topics: topicConfig.getStringArray('kafka.topics'), + }, + }; + }), + }; +}; + export const readConsumerConfig = ( config: Config, + logger: LoggerService, ): KafkaConsumingEventPublisherConfig[] => { - const publishers = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + const publishersConfig = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER); + + // Check for legacy single publisher format + if (publishersConfig?.getOptionalString('clientId')) { + logger.warn( + 'Legacy single config format detected at events.modules.kafka.kafkaConsumingEventPublisher.', + ); + return [ + processSinglePublisher( + 'default', // use `default` as instance name for legacy single config + publishersConfig, + ), + ]; + } return ( - publishers?.keys()?.map(publisherKey => { - const publisherConfig = publishers.getConfig(publisherKey); - - return { - instance: publisherKey, - kafkaConfig: readKafkaConfig(publisherConfig), - kafkaConsumerConfigs: publisherConfig - .getConfigArray('topics') - .map(topicConfig => { - return { - backstageTopic: topicConfig.getString('topic'), - consumerConfig: { - groupId: topicConfig.getString('kafka.groupId'), - sessionTimeout: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.sessionTimeout', - ), - rebalanceTimeout: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.rebalanceTimeout', - ), - heartbeatInterval: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.heartbeatInterval', - ), - metadataMaxAge: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.metadataMaxAge', - ), - maxBytesPerPartition: topicConfig.getOptionalNumber( - 'kafka.maxBytesPerPartition', - ), - minBytes: topicConfig.getOptionalNumber('kafka.minBytes'), - maxBytes: topicConfig.getOptionalNumber('kafka.maxBytes'), - maxWaitTimeInMs: readOptionalHumanDurationInMs( - topicConfig, - 'kafka.maxWaitTime', - ), - }, - consumerSubscribeTopics: { - topics: topicConfig.getStringArray('kafka.topics'), - }, - }; - }), - }; - }) ?? [] + publishersConfig + ?.keys() + ?.map(publisherKey => + processSinglePublisher( + publisherKey, + publishersConfig.getConfig(publisherKey), + ), + ) ?? [] ); };