Merge pull request #29315 from VELUX/feature/events-kafka-module
feat(events): add a new kafka module for events backend
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1,7 @@
|
||||
---
|
||||
'@backstage/plugin-events-backend-module-kafka': minor
|
||||
---
|
||||
|
||||
Adds a new module `kafka` for plugin-events-backend
|
||||
|
||||
The module introduces the `KafkaConsumerClient` which creates a Kafka client used to establish consumer connections. It also provides the `KafkaConsumingEventPublisher`, a consumer that subscribes to configured Kafka topics and publishes received messages to the Event Service.
|
||||
@@ -0,0 +1 @@
|
||||
module.exports = require('@backstage/cli/config/eslint-factory')(__dirname);
|
||||
@@ -0,0 +1,78 @@
|
||||
# @backstage/backstage-plugin-events-backend-module-kafka
|
||||
|
||||
Welcome to the `events-backend-module-kafka` backend module!
|
||||
|
||||
This package is a module for the `events-backend` backend plugin and extends the events system with an `KafkaConsumingEventPublisher.`
|
||||
|
||||
This event publisher will allow you to receive events from an Kafka queue and will publish these to the used `EventsService` implementation.
|
||||
|
||||
## Configuration
|
||||
|
||||
To set up Kafka queues, you need to configure the following values:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaConsumingEventPublisher:
|
||||
clientId: your-client-id # (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
|
||||
brokers: # (Required) List of brokers in the Kafka cluster to connect to.
|
||||
- broker1
|
||||
- broker2
|
||||
topics:
|
||||
- topic: 'backstage.topic' # (Required) Replace with actual topic name as expected by subscribers
|
||||
kafka:
|
||||
topics: # (Required) The Kafka topics to subscribe to.
|
||||
- topic1
|
||||
groupId: your-group-id # (Required) The GroupId to be used by the topic consumers.
|
||||
```
|
||||
|
||||
For a complete list of all available fields that can be configured, refer to the [config.d.ts file](./config.d.ts).
|
||||
|
||||
### Optional SSL Configuration
|
||||
|
||||
If your Kafka cluster requires SSL, you can configure it as follows:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaConsumingEventPublisher:
|
||||
ssl:
|
||||
rejectUnauthorized: true # (Optional) If true, the server certificate is verified against the list of supplied CAs.
|
||||
ca: [path/to/ca-cert] # (Optional) Array of trusted certificates in PEM format.
|
||||
key: path/to/client-key # (Optional) Private key in PEM format.
|
||||
cert: path/to/client-cert # (Optional) Public x509 certificate in PEM format.
|
||||
```
|
||||
|
||||
### Optional SASL Authentication Configuration
|
||||
|
||||
If your Kafka cluster requires `SASL` authentication, you can configure it as follows:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaConsumingEventPublisher:
|
||||
sasl:
|
||||
mechanism: 'plain' # SASL mechanism ('plain', 'scram-sha-256' or 'scram-sha-512')
|
||||
username: your-username # SASL username
|
||||
password: your-password # SASL password
|
||||
```
|
||||
|
||||
This section includes optional `SSL` and `SASL` authentication configuration for enhanced security.
|
||||
|
||||
## Installation
|
||||
|
||||
1. Install this module
|
||||
2. Add your configuration.
|
||||
|
||||
```bash
|
||||
# From your Backstage root directory
|
||||
yarn --cwd packages/backend add @backstage/plugin-events-backend-module-kafka
|
||||
```
|
||||
|
||||
```typescript
|
||||
// packages/backend/src/index.ts
|
||||
backend.add(import('@backstage/plugin-events-backend-module-kafka'));
|
||||
```
|
||||
@@ -0,0 +1,10 @@
|
||||
apiVersion: backstage.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: backstage-plugin-events-backend-module-kafka
|
||||
title: '@backstage/plugin-events-backend-module-kafka'
|
||||
description: The kafka backend module for the events plugin.
|
||||
spec:
|
||||
lifecycle: experimental
|
||||
type: backstage-backend-plugin-module
|
||||
owner: maintainers
|
||||
+205
@@ -0,0 +1,205 @@
|
||||
/*
|
||||
* 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 { HumanDuration } from '@backstage/types';
|
||||
|
||||
export interface Config {
|
||||
events?: {
|
||||
modules?: {
|
||||
/**
|
||||
* events-backend-module-kafka plugin configuration.
|
||||
*/
|
||||
kafka?: {
|
||||
/**
|
||||
* Configuration for KafkaConsumingEventPublisher
|
||||
*/
|
||||
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: {
|
||||
/**
|
||||
* (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 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: {
|
||||
/**
|
||||
* (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;
|
||||
};
|
||||
}>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
{
|
||||
"name": "@backstage/plugin-events-backend-module-kafka",
|
||||
"version": "0.0.0",
|
||||
"description": "The kafka backend module for the events plugin.",
|
||||
"backstage": {
|
||||
"role": "backend-plugin-module",
|
||||
"pluginId": "events",
|
||||
"pluginPackage": "@backstage/plugin-events-backend"
|
||||
},
|
||||
"publishConfig": {
|
||||
"access": "public",
|
||||
"main": "dist/index.cjs.js",
|
||||
"types": "dist/index.d.ts"
|
||||
},
|
||||
"homepage": "https://backstage.io",
|
||||
"repository": {
|
||||
"type": "git",
|
||||
"url": "https://github.com/backstage/backstage",
|
||||
"directory": "plugins/events-backend-module-kafka"
|
||||
},
|
||||
"license": "Apache-2.0",
|
||||
"main": "src/index.ts",
|
||||
"types": "src/index.ts",
|
||||
"files": [
|
||||
"dist",
|
||||
"config.d.ts"
|
||||
],
|
||||
"scripts": {
|
||||
"build": "backstage-cli package build",
|
||||
"clean": "backstage-cli package clean",
|
||||
"lint": "backstage-cli package lint",
|
||||
"prepack": "backstage-cli package prepack",
|
||||
"postpack": "backstage-cli package postpack",
|
||||
"start": "backstage-cli package start",
|
||||
"test": "backstage-cli package test"
|
||||
},
|
||||
"dependencies": {
|
||||
"@backstage/backend-plugin-api": "workspace:^",
|
||||
"@backstage/config": "workspace:^",
|
||||
"@backstage/plugin-events-node": "workspace:^",
|
||||
"@backstage/types": "workspace:^",
|
||||
"kafkajs": "^2.2.4",
|
||||
"luxon": "^3.0.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@backstage/backend-test-utils": "workspace:^",
|
||||
"@backstage/cli": "workspace:^",
|
||||
"@backstage/plugin-events-backend-test-utils": "workspace:^"
|
||||
},
|
||||
"configSchema": "config.d.ts"
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
## API Report File for "@backstage/plugin-events-backend-module-kafka"
|
||||
|
||||
> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
|
||||
|
||||
```ts
|
||||
import { BackendFeature } from '@backstage/backend-plugin-api';
|
||||
|
||||
// @public
|
||||
const eventsModuleKafkaConsumingEventPublisher: BackendFeature;
|
||||
export default eventsModuleKafkaConsumingEventPublisher;
|
||||
```
|
||||
@@ -0,0 +1,26 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*
|
||||
* @packageDocumentation
|
||||
*/
|
||||
|
||||
export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher';
|
||||
@@ -0,0 +1,132 @@
|
||||
/*
|
||||
* 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();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,79 @@
|
||||
/*
|
||||
* 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
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
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<void> {
|
||||
this.consumers.map(async consumer => await consumer.start());
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
this.consumers.map(async consumer => await consumer.shutdown());
|
||||
}
|
||||
}
|
||||
+92
@@ -0,0 +1,92 @@
|
||||
/*
|
||||
* 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();
|
||||
});
|
||||
});
|
||||
@@ -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 { 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.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
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 constructor(
|
||||
kafkaClient: Kafka,
|
||||
logger: LoggerService,
|
||||
private readonly events: EventsService,
|
||||
config: KafkaConsumerConfig,
|
||||
) {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
/*
|
||||
* 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 { LogEntry, logLevel } from 'kafkajs';
|
||||
|
||||
export const loggerServiceAdapter =
|
||||
(loggerService: LoggerService) => (_level: logLevel) => {
|
||||
return (entry: LogEntry) => {
|
||||
const { namespace, level, log } = entry;
|
||||
const { message, ...extra } = log;
|
||||
|
||||
const logMethods: Record<
|
||||
logLevel,
|
||||
(message: string, meta?: object) => void
|
||||
> = {
|
||||
[logLevel.ERROR]: loggerService.error,
|
||||
[logLevel.WARN]: loggerService.warn,
|
||||
[logLevel.INFO]: loggerService.info,
|
||||
[logLevel.DEBUG]: loggerService.debug,
|
||||
[logLevel.NOTHING]: () => {},
|
||||
};
|
||||
|
||||
// Use loggerService method that matches the level
|
||||
logMethods[level].call(
|
||||
loggerService,
|
||||
`Kafka ${namespace} ${log.message}`,
|
||||
{
|
||||
...extra,
|
||||
},
|
||||
);
|
||||
};
|
||||
};
|
||||
@@ -0,0 +1,250 @@
|
||||
/*
|
||||
* 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);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,159 @@
|
||||
/*
|
||||
* 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 { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs';
|
||||
import { Duration } from 'luxon';
|
||||
|
||||
/**
|
||||
* @public
|
||||
*/
|
||||
export interface KafkaConsumerConfig {
|
||||
backstageTopic: string;
|
||||
consumerConfig: ConsumerConfig;
|
||||
consumerSubscribeTopics: ConsumerSubscribeTopics;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
*/
|
||||
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 Duration.fromObject(humanDuration).as('milliseconds');
|
||||
};
|
||||
|
||||
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,
|
||||
};
|
||||
};
|
||||
+85
@@ -0,0 +1,85 @@
|
||||
/*
|
||||
* 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();
|
||||
});
|
||||
});
|
||||
+56
@@ -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 { KafkaConsumerClient } from '../publisher/KafkaConsumerClient';
|
||||
import { eventsServiceRef } from '@backstage/plugin-events-node';
|
||||
|
||||
/**
|
||||
* Kafka module for the Events plugin.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export const eventsModuleKafkaConsumingEventPublisher = createBackendModule({
|
||||
pluginId: 'events',
|
||||
moduleId: 'kafka-consuming-event-publisher',
|
||||
register(env) {
|
||||
env.registerInit({
|
||||
deps: {
|
||||
config: coreServices.rootConfig,
|
||||
events: eventsServiceRef,
|
||||
logger: coreServices.logger,
|
||||
lifecycle: coreServices.lifecycle,
|
||||
},
|
||||
async init({ config, logger, events, lifecycle }) {
|
||||
const kafka = KafkaConsumerClient.fromConfig({
|
||||
config,
|
||||
events,
|
||||
logger,
|
||||
});
|
||||
|
||||
if (!kafka) {
|
||||
return;
|
||||
}
|
||||
|
||||
await kafka.start();
|
||||
|
||||
lifecycle.addShutdownHook(async () => await kafka.shutdown());
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -6700,6 +6700,22 @@ __metadata:
|
||||
languageName: unknown
|
||||
linkType: soft
|
||||
|
||||
"@backstage/plugin-events-backend-module-kafka@workspace:plugins/events-backend-module-kafka":
|
||||
version: 0.0.0-use.local
|
||||
resolution: "@backstage/plugin-events-backend-module-kafka@workspace:plugins/events-backend-module-kafka"
|
||||
dependencies:
|
||||
"@backstage/backend-plugin-api": "workspace:^"
|
||||
"@backstage/backend-test-utils": "workspace:^"
|
||||
"@backstage/cli": "workspace:^"
|
||||
"@backstage/config": "workspace:^"
|
||||
"@backstage/plugin-events-backend-test-utils": "workspace:^"
|
||||
"@backstage/plugin-events-node": "workspace:^"
|
||||
"@backstage/types": "workspace:^"
|
||||
kafkajs: "npm:^2.2.4"
|
||||
luxon: "npm:^3.0.0"
|
||||
languageName: unknown
|
||||
linkType: soft
|
||||
|
||||
"@backstage/plugin-events-backend-test-utils@workspace:^, @backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils":
|
||||
version: 0.0.0-use.local
|
||||
resolution: "@backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils"
|
||||
@@ -35319,6 +35335,13 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"kafkajs@npm:^2.2.4":
|
||||
version: 2.2.4
|
||||
resolution: "kafkajs@npm:2.2.4"
|
||||
checksum: 10/75eb0d221397085f90e51f8a2d752495c9fa9c1b3a1a6db610cd7074fa8c52777f295832fd0a7c49cded5e574337a09fafa8c3f7cf1caa38f4dc9aa20fcfb7df
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"keygrip@npm:~1.1.0":
|
||||
version: 1.1.0
|
||||
resolution: "keygrip@npm:1.1.0"
|
||||
|
||||
Reference in New Issue
Block a user