From c7065e4c266dcc2a89923e5ced221b708b6027b6 Mon Sep 17 00:00:00 2001 From: Jonas Beck Date: Wed, 4 Jun 2025 12:30:30 +0200 Subject: [PATCH] chore(events): use readDurationFromConfig for all config durations Signed-off-by: Jonas Beck --- .../events-backend-module-kafka/config.d.ts | 40 +++++----- .../events-backend-module-kafka/package.json | 3 +- .../src/publisher/config.test.ts | 80 +++++++++++++++---- .../src/publisher/config.ts | 77 +++++++++++++++--- yarn.lock | 1 + 5 files changed, 155 insertions(+), 46 deletions(-) diff --git a/plugins/events-backend-module-kafka/config.d.ts b/plugins/events-backend-module-kafka/config.d.ts index 4c0d06c77e..c3400d43cd 100644 --- a/plugins/events-backend-module-kafka/config.d.ts +++ b/plugins/events-backend-module-kafka/config.d.ts @@ -60,16 +60,16 @@ export interface Config { */ retry: { /** - * (Optional) Maximum wait time for a retry in milliseconds + * (Optional) Maximum wait time for a retry * Default: 30000 ms. */ - maxRetryTime: number; + maxRetryTime: HumanDuration | string; /** - * (Optional) Initial value used to calculate the retry in milliseconds (This is still randomized following the randomization factor) + * (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor) * Default: 300 ms. */ - initialRetryTime: number; + initialRetryTime: HumanDuration | string; /** * (Optional) Randomization factor @@ -91,22 +91,22 @@ export interface Config { }; /** - * (Optional) Timeout in ms for authentication requests. + * (Optional) Timeout for authentication requests. * Default: 10000 ms. */ - authenticationTimeout: number; + authenticationTimeout: HumanDuration | string; /** - * (Optional) Time in milliseconds to wait for a successful connection. + * (Optional) Time to wait for a successful connection. * Default: 1000 ms. */ - connectionTimeout: number; + connectionTimeout: HumanDuration | string; /** - * (Optional) Time in milliseconds to wait for a successful request. + * (Optional) Time to wait for a successful request. * Default: 30000 ms. */ - requestTimeout: number; + requestTimeout: HumanDuration | string; /** * (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false. @@ -137,34 +137,34 @@ export interface Config { groupId: string; /** - * (Optional) Timeout in milliseconds used to detect failures. + * (Optional) Timeout used to detect failures. * The consumer sends periodic heartbeats to indicate its liveness to the broker. * If no heartbeats are received by the broker before the expiration of this session timeout, * then the broker will remove this consumer from the group and initiate a rebalance * Default: 30000 ms. */ - sessionTimeout: number; + sessionTimeout: HumanDuration | string; /** * (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group * Default: 60000 ms. */ - rebalanceTimeout: number; + rebalanceTimeout: HumanDuration | string; /** - * (Optional) The expected time in milliseconds between heartbeats to the consumer coordinator. + * (Optional) The expected time between heartbeats to the consumer coordinator. * Heartbeats are used to ensure that the consumer's session stays active. * The value must be set lower than session timeout * Default: 3000 ms. */ - heartbeatInterval: number; + heartbeatInterval: HumanDuration | string; /** - * (Optional) The period of time in milliseconds after which we force a refresh of metadata + * (Optional) The period of time after which we force a refresh of metadata * even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions * Default: 300000 ms (5 minutes). */ - metadataMaxAge: number; + metadataMaxAge: HumanDuration | string; /** * (Optional) The maximum amount of data per-partition the server will return. @@ -176,7 +176,7 @@ export interface Config { maxBytesPerPartition: number; /** - * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTimeInMs for more data to accumulate. + * (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate. * Default: 1 */ minBytes: number; @@ -188,11 +188,11 @@ export interface Config { maxBytes: number; /** - * (Optional) The maximum amount of time in milliseconds the server will block before answering the fetch request + * (Optional) The maximum amount of time the server will block before answering the fetch request * if there isn’t sufficient data to immediately satisfy the requirement given by minBytes * Default: 5000 */ - maxWaitTimeInMs: number; + maxWaitTime: HumanDuration | string; }; }>; }; diff --git a/plugins/events-backend-module-kafka/package.json b/plugins/events-backend-module-kafka/package.json index 4b9594c4af..488ecc0a0e 100644 --- a/plugins/events-backend-module-kafka/package.json +++ b/plugins/events-backend-module-kafka/package.json @@ -39,7 +39,8 @@ "@backstage/config": "workspace:^", "@backstage/plugin-events-node": "workspace:^", "@backstage/types": "workspace:^", - "kafkajs": "^2.2.4" + "kafkajs": "^2.2.4", + "luxon": "^3.0.0" }, "devDependencies": { "@backstage/backend-test-utils": "workspace:^", diff --git a/plugins/events-backend-module-kafka/src/publisher/config.test.ts b/plugins/events-backend-module-kafka/src/publisher/config.test.ts index 5b2dfb6f0c..fc7e3049b0 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.test.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.test.ts @@ -90,15 +90,15 @@ describe('readConfig', () => { password: 'password', }, retry: { - maxRetryTime: '20000', - initialRetryTime: '200', + maxRetryTime: { milliseconds: 20000 }, + initialRetryTime: { milliseconds: 200 }, factor: '0.4', multiplier: '4', retries: '10', }, - authenticationTimeout: 20000, - connectionTimeout: 1500, - requestTimeout: 20000, + authenticationTimeout: { milliseconds: 20000 }, + connectionTimeout: { milliseconds: 1500 }, + requestTimeout: { milliseconds: 20000 }, enforceRequestTimeout: false, topics: [ { @@ -106,14 +106,14 @@ describe('readConfig', () => { kafka: { topics: ['topic-A'], groupId: 'my-group', - sessionTimeout: 20000, - rebalanceTimeout: 50000, - heartbeatInterval: 2000, - metadataMaxAge: 400000, + sessionTimeout: { milliseconds: 20000 }, + rebalanceTimeout: { milliseconds: 50000 }, + heartbeatInterval: { milliseconds: 2000 }, + metadataMaxAge: { milliseconds: 400000 }, maxBytesPerPartition: 50000, minBytes: 2, maxBytes: 500000, - maxWaitTimeInMs: 4000, + maxWaitTime: { milliseconds: 4000 }, }, }, { @@ -151,11 +151,11 @@ describe('readConfig', () => { expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(20000); expect(publisherConfigs?.kafkaConfig.enforceRequestTimeout).toBeFalsy(); expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({ - maxRetryTime: '20000', - initialRetryTime: '200', - factor: '0.4', - multiplier: '4', - retries: '10', + maxRetryTime: 20000, + initialRetryTime: 200, + factor: 0.4, + multiplier: 4, + retries: 10, }); // Consumer configuration @@ -197,4 +197,54 @@ describe('readConfig', () => { publisherConfigs?.kafkaConsumerConfigs[0].consumerConfig.maxWaitTimeInMs, ).toBe(4000); }); + + it('should handle HumanDuration and string values for durations and timeouts', () => { + const config = new ConfigReader({ + events: { + modules: { + kafka: { + kafkaConsumingEventPublisher: { + clientId: 'backstage-events', + brokers: ['kafka1:9092', 'kafka2:9092'], + retry: { + maxRetryTime: { seconds: 1 }, + initialRetryTime: { minutes: 1 }, + factor: 0.4, + multiplier: 4, + retries: 10, + }, + authenticationTimeout: { hours: 1 }, + connectionTimeout: { days: 1 }, + topics: [], + requestTimeout: '1m', + }, + }, + }, + }, + }); + + const publisherConfigs = readConfig(config); + + expect(publisherConfigs).toBeDefined(); + + // Client configuration + expect(publisherConfigs?.kafkaConfig.clientId).toEqual('backstage-events'); + expect(publisherConfigs?.kafkaConfig.brokers).toEqual([ + 'kafka1:9092', + 'kafka2:9092', + ]); + expect(publisherConfigs?.kafkaConfig.authenticationTimeout).toBe(3600000); + expect(publisherConfigs?.kafkaConfig.connectionTimeout).toBe(86400000); + expect(publisherConfigs?.kafkaConfig.requestTimeout).toBe(60000); + expect(publisherConfigs?.kafkaConfig.retry).toStrictEqual({ + maxRetryTime: 1000, + initialRetryTime: 60000, + factor: 0.4, + multiplier: 4, + retries: 10, + }); + + // Consumer configuration + expect(publisherConfigs?.kafkaConsumerConfigs.length).toBe(0); + }); }); diff --git a/plugins/events-backend-module-kafka/src/publisher/config.ts b/plugins/events-backend-module-kafka/src/publisher/config.ts index aadf4a5166..aa5921a7f3 100644 --- a/plugins/events-backend-module-kafka/src/publisher/config.ts +++ b/plugins/events-backend-module-kafka/src/publisher/config.ts @@ -13,8 +13,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { Config } from '@backstage/config'; +import { Config, readDurationFromConfig } from '@backstage/config'; import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs'; +import { Duration } from 'luxon'; /** * @public @@ -36,6 +37,26 @@ export interface KafkaEventSourceConfig { const CONFIG_PREFIX_PUBLISHER = 'events.modules.kafka.kafkaConsumingEventPublisher'; +/** + * Reads an optional HumanDuration from the config and returns the value in milliseconds if the key is defined. + * + * @param config - The configuration object to read from. + * @param key - The key to look up in the configuration. + * @returns The duration in milliseconds, or undefined if the key is not defined. + */ +const readOptionalHumanDurationInMs = ( + config: Config, + key: string, +): number | undefined => { + const humanDuration = config.has(key) + ? readDurationFromConfig(config, { key }) + : undefined; + + if (!humanDuration) return undefined; + + return Duration.fromObject(humanDuration).as('milliseconds'); +}; + export const readConfig = ( config: Config, ): KafkaEventSourceConfig | undefined => { @@ -48,18 +69,39 @@ export const readConfig = ( const clientId = kafkaConfig.getString('clientId'); const brokers = kafkaConfig.getStringArray('brokers'); - const authenticationTimeout = kafkaConfig.getOptionalNumber( + const authenticationTimeout = readOptionalHumanDurationInMs( + kafkaConfig, 'authenticationTimeout', ); - const connectionTimeout = kafkaConfig.getOptionalNumber('connectionTimeout'); - const requestTimeout = kafkaConfig.getOptionalNumber('requestTimeout'); + + const connectionTimeout = readOptionalHumanDurationInMs( + kafkaConfig, + 'connectionTimeout', + ); + const requestTimeout = readOptionalHumanDurationInMs( + kafkaConfig, + 'requestTimeout', + ); const enforceRequestTimeout = kafkaConfig.getOptionalBoolean( 'enforceRequestTimeout', ); const ssl = kafkaConfig.getOptional('ssl') as KafkaConfig['ssl']; const sasl = kafkaConfig.getOptional('sasl') as KafkaConfig['sasl']; - const retry = kafkaConfig.getOptional('retry') as KafkaConfig['retry']; + + const retry: KafkaConfig['retry'] = { + maxRetryTime: readOptionalHumanDurationInMs( + kafkaConfig, + 'retry.maxRetryTime', + ), + initialRetryTime: readOptionalHumanDurationInMs( + kafkaConfig, + 'retry.initialRetryTime', + ), + factor: kafkaConfig.getOptionalNumber('retry.factor'), + multiplier: kafkaConfig.getOptionalNumber('retry.multiplier'), + retries: kafkaConfig.getOptionalNumber('retry.retries'), + }; const kafkaConsumerConfigs: KafkaConsumerConfig[] = kafkaConfig .getConfigArray('topics') @@ -68,16 +110,31 @@ export const readConfig = ( backstageTopic: topic.getString('topic'), consumerConfig: { groupId: topic.getString('kafka.groupId'), - sessionTimeout: topic.getOptionalNumber('kafka.sessionTimeout'), - rebalanceTimeout: topic.getOptionalNumber('kafka.rebalanceTimeout'), - heartbeatInterval: topic.getOptionalNumber('kafka.heartbeatInterval'), - metadataMaxAge: topic.getOptionalNumber('kafka.metadataMaxAge'), + sessionTimeout: readOptionalHumanDurationInMs( + topic, + 'kafka.sessionTimeout', + ), + rebalanceTimeout: readOptionalHumanDurationInMs( + topic, + 'kafka.rebalanceTimeout', + ), + heartbeatInterval: readOptionalHumanDurationInMs( + topic, + 'kafka.heartbeatInterval', + ), + metadataMaxAge: readOptionalHumanDurationInMs( + topic, + 'kafka.metadataMaxAge', + ), maxBytesPerPartition: topic.getOptionalNumber( 'kafka.maxBytesPerPartition', ), minBytes: topic.getOptionalNumber('kafka.minBytes'), maxBytes: topic.getOptionalNumber('kafka.maxBytes'), - maxWaitTimeInMs: topic.getOptionalNumber('kafka.maxWaitTimeInMs'), + maxWaitTimeInMs: readOptionalHumanDurationInMs( + topic, + 'kafka.maxWaitTime', + ), }, consumerSubscribeTopics: { topics: topic.getStringArray('kafka.topics'), diff --git a/yarn.lock b/yarn.lock index 11e3bd9ae4..6cadbc2dfd 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6647,6 +6647,7 @@ __metadata: "@backstage/plugin-events-node": "workspace:^" "@backstage/types": "workspace:^" kafkajs: "npm:^2.2.4" + luxon: "npm:^3.0.0" languageName: unknown linkType: soft