feat(events): add events kafka module
Signed-off-by: Jonas Beck <jonas.beck@velux.com>
This commit is contained in:
Jonas Beck
@@ -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](path/to/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
|
||||
+202
@@ -0,0 +1,202 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
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 in milliseconds
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
maxRetryTime: number;
|
||||
|
||||
/**
|
||||
* (Optional) Initial value used to calculate the retry in milliseconds (This is still randomized following the randomization factor)
|
||||
* Default: 300 ms.
|
||||
*/
|
||||
initialRetryTime: number;
|
||||
|
||||
/**
|
||||
* (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 in ms for authentication requests.
|
||||
* Default: 10000 ms.
|
||||
*/
|
||||
authenticationTimeout: number;
|
||||
|
||||
/**
|
||||
* (Optional) Time in milliseconds to wait for a successful connection.
|
||||
* Default: 1000 ms.
|
||||
*/
|
||||
connectionTimeout: number;
|
||||
|
||||
/**
|
||||
* (Optional) Time in milliseconds to wait for a successful request.
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
requestTimeout: number;
|
||||
|
||||
/**
|
||||
* (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 in milliseconds 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;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
|
||||
* Default: 60000 ms.
|
||||
*/
|
||||
rebalanceTimeout: number;
|
||||
|
||||
/**
|
||||
* (Optional) The expected time in milliseconds 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;
|
||||
|
||||
/**
|
||||
* (Optional) The period of time in milliseconds 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;
|
||||
|
||||
/**
|
||||
* (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 maxWaitTimeInMs 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 in milliseconds 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;
|
||||
};
|
||||
}>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"name": "@backstage/plugin-events-backend-module-kafka",
|
||||
"description": "The kafka backend module for the events plugin.",
|
||||
"version": "0.0.0",
|
||||
"main": "src/index.ts",
|
||||
"types": "src/index.ts",
|
||||
"license": "Apache-2.0",
|
||||
"private": true,
|
||||
"publishConfig": {
|
||||
"access": "public",
|
||||
"main": "dist/index.cjs.js",
|
||||
"types": "dist/index.d.ts"
|
||||
},
|
||||
"backstage": {
|
||||
"role": "backend-plugin-module"
|
||||
},
|
||||
"scripts": {
|
||||
"start": "backstage-cli package start",
|
||||
"build": "backstage-cli package build",
|
||||
"lint": "backstage-cli package lint",
|
||||
"test": "backstage-cli package test",
|
||||
"clean": "backstage-cli package clean",
|
||||
"prepack": "backstage-cli package prepack",
|
||||
"postpack": "backstage-cli package postpack"
|
||||
},
|
||||
"dependencies": {
|
||||
"@backstage/backend-plugin-api": "workspace:^",
|
||||
"@backstage/config": "workspace:^",
|
||||
"@backstage/plugin-events-node": "workspace:^",
|
||||
"@backstage/types": "workspace:^",
|
||||
"kafkajs": "^2.2.4"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@backstage/backend-test-utils": "workspace:^",
|
||||
"@backstage/cli": "workspace:^",
|
||||
"@backstage/plugin-events-backend-test-utils": "workspace:^"
|
||||
},
|
||||
"files": [
|
||||
"dist"
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
## API Report File for "@internal/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';
|
||||
import { Config } from '@backstage/config';
|
||||
import { ConsumerConfig } from 'kafkajs';
|
||||
import { ConsumerSubscribeTopics } from 'kafkajs';
|
||||
import { EventsService } from '@backstage/plugin-events-node';
|
||||
import { Kafka } from 'kafkajs';
|
||||
import { LoggerService } from '@backstage/backend-plugin-api';
|
||||
|
||||
// @public
|
||||
const eventsModuleKafkaConsumingEventPublisher: BackendFeature;
|
||||
export default eventsModuleKafkaConsumingEventPublisher;
|
||||
|
||||
// @public (undocumented)
|
||||
export class KafkaConsumerClient {
|
||||
// (undocumented)
|
||||
static fromConfig(env: {
|
||||
config: Config;
|
||||
events: EventsService;
|
||||
logger: LoggerService;
|
||||
}): KafkaConsumerClient;
|
||||
// (undocumented)
|
||||
shutdown(): Promise<void>;
|
||||
// (undocumented)
|
||||
start(): Promise<void>;
|
||||
}
|
||||
|
||||
// @public (undocumented)
|
||||
export class KafkaConsumingEventPublisher {
|
||||
// (undocumented)
|
||||
static fromConfig(env: {
|
||||
kafkaClient: Kafka;
|
||||
config: KafkaConsumerConfig;
|
||||
events: EventsService;
|
||||
logger: LoggerService;
|
||||
}): KafkaConsumingEventPublisher;
|
||||
// (undocumented)
|
||||
shutdown(): Promise<void>;
|
||||
// (undocumented)
|
||||
start(): Promise<void>;
|
||||
}
|
||||
|
||||
// Warnings were encountered during analysis:
|
||||
//
|
||||
// src/publisher/KafkaConsumingEventPublisher.d.ts:13:9 - (ae-forgotten-export) The symbol "KafkaConsumerConfig" needs to be exported by the entry point index.d.ts
|
||||
```
|
||||
@@ -0,0 +1,19 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
export { KafkaConsumerClient } from './publisher/KafkaConsumerClient';
|
||||
export { KafkaConsumingEventPublisher } from './publisher/KafkaConsumingEventPublisher';
|
||||
export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher';
|
||||
@@ -0,0 +1,118 @@
|
||||
/*
|
||||
* 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 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,
|
||||
});
|
||||
|
||||
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,
|
||||
});
|
||||
|
||||
await client.shutdown();
|
||||
|
||||
expect(mockConsumer.shutdown).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,67 @@
|
||||
/*
|
||||
* 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 { LoggerServiceCreator } from './LoggerServiceCreator';
|
||||
|
||||
export class KafkaConsumerClient {
|
||||
private readonly kafka: Kafka;
|
||||
private readonly consumers: KafkaConsumingEventPublisher[];
|
||||
|
||||
static fromConfig(env: {
|
||||
config: Config;
|
||||
events: EventsService;
|
||||
logger: LoggerService;
|
||||
}): KafkaConsumerClient {
|
||||
return new KafkaConsumerClient(
|
||||
env.logger,
|
||||
env.events,
|
||||
readConfig(env.config),
|
||||
);
|
||||
}
|
||||
|
||||
private constructor(
|
||||
logger: LoggerService,
|
||||
events: EventsService,
|
||||
config: KafkaEventSourceConfig,
|
||||
) {
|
||||
this.kafka = new Kafka({
|
||||
...config.kafkaConfig,
|
||||
logCreator: LoggerServiceCreator(logger),
|
||||
});
|
||||
|
||||
this.consumers = config.kafkaConsumerConfig.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',
|
||||
},
|
||||
consumerSubscribeConfig: {
|
||||
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.consumerSubscribeConfig,
|
||||
);
|
||||
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,104 @@
|
||||
/*
|
||||
* 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'];
|
||||
|
||||
export class KafkaConsumingEventPublisher {
|
||||
private readonly kafkaConsumer: Consumer;
|
||||
private readonly consumerSubscribeOptions: 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.consumerSubscribeOptions = config.consumerSubscribeConfig;
|
||||
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.consumerSubscribeConfig.topics.toString(),
|
||||
backstageTopic: config.backstageTopic,
|
||||
taskId: id,
|
||||
});
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
try {
|
||||
await this.kafkaConsumer.connect();
|
||||
this.logger.info('Kafka consumer connected');
|
||||
|
||||
await this.kafkaConsumer.subscribe(this.consumerSubscribeOptions);
|
||||
|
||||
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();
|
||||
this.logger.info('Kafka consumer disconnected');
|
||||
}
|
||||
|
||||
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 LoggerServiceCreator =
|
||||
(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,197 @@
|
||||
/*
|
||||
* 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 config = new ConfigReader({});
|
||||
|
||||
expect(() => {
|
||||
readConfig(config);
|
||||
}).toThrow();
|
||||
});
|
||||
|
||||
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.kafkaConsumerConfig.length).toBe(2);
|
||||
|
||||
expect(publisherConfigs.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(publisherConfigs.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual(
|
||||
'fake1',
|
||||
);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId,
|
||||
).toEqual('my-group');
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.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: '20000',
|
||||
initialRetryTime: '200',
|
||||
factor: '0.4',
|
||||
multiplier: '4',
|
||||
retries: '10',
|
||||
},
|
||||
authenticationTimeout: 20000,
|
||||
connectionTimeout: 1500,
|
||||
requestTimeout: 20000,
|
||||
enforceRequestTimeout: false,
|
||||
topics: [
|
||||
{
|
||||
topic: 'fake1',
|
||||
kafka: {
|
||||
topics: ['topic-A'],
|
||||
groupId: 'my-group',
|
||||
sessionTimeout: 20000,
|
||||
rebalanceTimeout: 50000,
|
||||
heartbeatInterval: 2000,
|
||||
metadataMaxAge: 400000,
|
||||
maxBytesPerPartition: 50000,
|
||||
minBytes: 2,
|
||||
maxBytes: 500000,
|
||||
maxWaitTimeInMs: 4000,
|
||||
},
|
||||
},
|
||||
{
|
||||
topic: 'fake2',
|
||||
kafka: {
|
||||
topics: ['topic-B'],
|
||||
groupId: 'my-group',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readConfig(config);
|
||||
|
||||
// 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.kafkaConsumerConfig.length).toBe(2);
|
||||
expect(publisherConfigs.kafkaConsumerConfig[0].backstageTopic).toEqual(
|
||||
'fake1',
|
||||
);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.groupId,
|
||||
).toEqual('my-group');
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerSubscribeConfig.topics,
|
||||
).toEqual(['topic-A']);
|
||||
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.sessionTimeout,
|
||||
).toBe(20000);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.rebalanceTimeout,
|
||||
).toBe(50000);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.heartbeatInterval,
|
||||
).toBe(2000);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.metadataMaxAge,
|
||||
).toBe(400000);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig
|
||||
.maxBytesPerPartition,
|
||||
).toBe(50000);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.minBytes,
|
||||
).toBe(2);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxBytes,
|
||||
).toBe(500000);
|
||||
expect(
|
||||
publisherConfigs.kafkaConsumerConfig[0].consumerConfig.maxWaitTimeInMs,
|
||||
).toBe(4000);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,90 @@
|
||||
/*
|
||||
* 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 } from '@backstage/config';
|
||||
import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs';
|
||||
|
||||
export interface KafkaConsumerConfig {
|
||||
backstageTopic: string;
|
||||
consumerConfig: ConsumerConfig;
|
||||
consumerSubscribeConfig: ConsumerSubscribeTopics;
|
||||
}
|
||||
|
||||
export interface KafkaEventSourceConfig {
|
||||
kafkaConfig: KafkaConfig;
|
||||
kafkaConsumerConfig: KafkaConsumerConfig[];
|
||||
}
|
||||
|
||||
const CONFIG_PREFIX_PUBLISHER =
|
||||
'events.modules.kafka.kafkaConsumingEventPublisher';
|
||||
|
||||
export const readConfig = (config: Config): KafkaEventSourceConfig => {
|
||||
const kafkaConfig = config.getConfig(CONFIG_PREFIX_PUBLISHER);
|
||||
|
||||
const clientId = kafkaConfig.getString('clientId');
|
||||
const brokers = kafkaConfig.getStringArray('brokers');
|
||||
|
||||
const authenticationTimeout = kafkaConfig.getOptionalNumber(
|
||||
'authenticationTimeout',
|
||||
);
|
||||
const connectionTimeout = kafkaConfig.getOptionalNumber('connectionTimeout');
|
||||
const requestTimeout = kafkaConfig.getOptionalNumber('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 kafkaConsumerConfig: KafkaConsumerConfig[] = kafkaConfig
|
||||
.getConfigArray('topics')
|
||||
.map(topic => {
|
||||
return {
|
||||
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'),
|
||||
maxBytesPerPartition: topic.getOptionalNumber(
|
||||
'kafka.maxBytesPerPartition',
|
||||
),
|
||||
minBytes: topic.getOptionalNumber('kafka.minBytes'),
|
||||
maxBytes: topic.getOptionalNumber('kafka.maxBytes'),
|
||||
maxWaitTimeInMs: topic.getOptionalNumber('kafka.maxWaitTimeInMs'),
|
||||
},
|
||||
consumerSubscribeConfig: {
|
||||
topics: topic.getStringArray('kafka.topics'),
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
return {
|
||||
kafkaConfig: {
|
||||
clientId,
|
||||
brokers,
|
||||
ssl,
|
||||
sasl,
|
||||
authenticationTimeout,
|
||||
connectionTimeout,
|
||||
requestTimeout,
|
||||
enforceRequestTimeout,
|
||||
retry,
|
||||
},
|
||||
kafkaConsumerConfig,
|
||||
};
|
||||
};
|
||||
+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();
|
||||
});
|
||||
});
|
||||
+52
@@ -0,0 +1,52 @@
|
||||
/*
|
||||
* 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,
|
||||
});
|
||||
|
||||
await kafka.start();
|
||||
|
||||
lifecycle.addShutdownHook(async () => await kafka.shutdown());
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -35254,6 +35254,13 @@ __metadata:
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"kafkajs@npm:^2.2.4":
|
||||
version: 2.2.4
|
||||
resolution: "kafkajs@npm:2.2.4"
|
||||
checksum: 83e9e8bc50a09b142f4ff79f6a2bd88ecc21b83bcefe6621ab1716118d624886befb7371731274f67812ce35dd50b53140ff3b49a06e5d9169fe6b164d72fea5
|
||||
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