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

chore(events): use readDurationFromConfig for all config durations

Browse Source 
Signed-off-by: Jonas Beck <jonas.beck@velux.com>
This commit is contained in:
Jonas Beck
2025-06-04 12:30:30 +02:00
parent 9093d77867
commit c7065e4c26
5 changed files with 155 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/events-backend-module-kafka/config.d.ts
Vendored
+20 -20
View File
@@ -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;
};
}>;
};
plugins/events-backend-module-kafka/package.json
+2 -1
View File
@@ -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:^",
plugins/events-backend-module-kafka/src/publisher/config.test.ts
+65 -15
View File
@@ -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);
});
});
plugins/events-backend-module-kafka/src/publisher/config.ts
+67 -10
View File
@@ -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'),
yarn.lock
+1
View File
@@ -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