paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 22 Packages Projects Releases Wiki Activity

feat(events): add backwards compatability to kafkaConsumingEventPublisher config

Browse Source 
Signed-off-by: Jonas Beck <dev@jonasbeck.dk>
This commit is contained in:
Jonas Beck
2025-12-11 10:16:46 +01:00
parent d3d33a0870
commit 9fa23b3ea1
5 changed files with 488 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/angry-views-win.md
+1 -19
View File
@@ -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.
plugins/events-backend-module-kafka/config.d.ts
Vendored
+331 -156
View File
@@ -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
plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/KafkaConsumingEventPublisher.ts
+1 -1
View File
@@ -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 =>
plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.test.ts
+83 -4
View File
@@ -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.',
);
});
});
plugins/events-backend-module-kafka/src/KafkaConsumingEventPublisher/config.ts
+72 -47
View File
@@ -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),
),
) ?? []
);
};