feat(events): restructure kafka module and add publisher
Signed-off-by: Jonas Beck <dev@jonasbeck.dk>
This commit is contained in:
Jonas Beck
+344
-151
@@ -13,7 +13,6 @@
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { HumanDuration } from '@backstage/types';
|
||||
|
||||
export interface Config {
|
||||
@@ -25,179 +24,373 @@ export interface Config {
|
||||
kafka?: {
|
||||
/**
|
||||
* Configuration for KafkaConsumingEventPublisher
|
||||
*
|
||||
* Supports multiple named instances as a record where each key is a unique name
|
||||
* for the Kafka consumer configuration.
|
||||
*/
|
||||
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: {
|
||||
[name: string]: {
|
||||
/**
|
||||
* (Optional) Maximum wait time for a retry
|
||||
* Default: 30000 ms.
|
||||
* (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
|
||||
*/
|
||||
maxRetryTime: HumanDuration | string;
|
||||
clientId: string;
|
||||
/**
|
||||
* (Required) List of brokers in the Kafka cluster to connect to.
|
||||
*/
|
||||
brokers: string[];
|
||||
/**
|
||||
* Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect.
|
||||
* See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options
|
||||
*/
|
||||
ssl?:
|
||||
| {
|
||||
ca?: string[];
|
||||
/** @visibility secret */
|
||||
key?: string;
|
||||
cert?: string;
|
||||
rejectUnauthorized?: boolean;
|
||||
}
|
||||
| boolean;
|
||||
/**
|
||||
* Optional SASL connection parameters.
|
||||
*/
|
||||
sasl?: {
|
||||
mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
|
||||
username: string;
|
||||
/** @visibility secret */
|
||||
password: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor)
|
||||
* Default: 300 ms.
|
||||
* Optional retry connection parameters.
|
||||
*/
|
||||
initialRetryTime: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Randomization factor
|
||||
* Default: 0.2.
|
||||
*/
|
||||
factor: number;
|
||||
|
||||
/**
|
||||
* (Optional) Exponential factor
|
||||
* Default: 2.
|
||||
*/
|
||||
multiplier: number;
|
||||
|
||||
/**
|
||||
* (Optional) Max number of retries per call
|
||||
* Default: 5.
|
||||
*/
|
||||
retries: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* (Optional) Timeout for authentication requests.
|
||||
* Default: 10000 ms.
|
||||
*/
|
||||
authenticationTimeout: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful connection.
|
||||
* Default: 1000 ms.
|
||||
*/
|
||||
connectionTimeout: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful request.
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
requestTimeout: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false.
|
||||
* Default: true
|
||||
*/
|
||||
enforceRequestTimeout: boolean;
|
||||
|
||||
/**
|
||||
* Contains 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: {
|
||||
retry?: {
|
||||
/**
|
||||
* (Required) The Kafka topics to subscribe to
|
||||
*/
|
||||
topics: string[];
|
||||
/**
|
||||
* (Required) The GroupId to be used by the topic consumers
|
||||
*/
|
||||
groupId: string;
|
||||
|
||||
/**
|
||||
* (Optional) Timeout used to detect failures.
|
||||
* The consumer sends periodic heartbeats to indicate its liveness to the broker.
|
||||
* If no heartbeats are received by the broker before the expiration of this session timeout,
|
||||
* then the broker will remove this consumer from the group and initiate a rebalance
|
||||
* (Optional) Maximum wait time for a retry
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
sessionTimeout: HumanDuration | string;
|
||||
maxRetryTime?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
|
||||
* Default: 60000 ms.
|
||||
* (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor)
|
||||
* Default: 300 ms.
|
||||
*/
|
||||
rebalanceTimeout: HumanDuration | string;
|
||||
initialRetryTime?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The expected time between heartbeats to the consumer coordinator.
|
||||
* Heartbeats are used to ensure that the consumer's session stays active.
|
||||
* The value must be set lower than session timeout
|
||||
* Default: 3000 ms.
|
||||
* (Optional) Randomization factor
|
||||
* Default: 0.2.
|
||||
*/
|
||||
heartbeatInterval: HumanDuration | string;
|
||||
factor?: number;
|
||||
|
||||
/**
|
||||
* (Optional) The period of time after which we force a refresh of metadata
|
||||
* even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions
|
||||
* Default: 300000 ms (5 minutes).
|
||||
* (Optional) Exponential factor
|
||||
* Default: 2.
|
||||
*/
|
||||
metadataMaxAge: HumanDuration | string;
|
||||
multiplier?: number;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum amount of data per-partition the server will return.
|
||||
* This size must be at least as large as the maximum message size the server allows
|
||||
* or else it is possible for the producer to send messages larger than the consumer can fetch.
|
||||
* If that happens, the consumer can get stuck trying to fetch a large message on a certain partition
|
||||
* Default: 1048576 (1MB)
|
||||
* (Optional) Max number of retries per call
|
||||
* Default: 5.
|
||||
*/
|
||||
maxBytesPerPartition: number;
|
||||
|
||||
/**
|
||||
* (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate.
|
||||
* Default: 1
|
||||
*/
|
||||
minBytes: number;
|
||||
|
||||
/**
|
||||
* (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0
|
||||
* Default: 10485760 (10MB)
|
||||
*/
|
||||
maxBytes: number;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum amount of time the server will block before answering the fetch request
|
||||
* if there isn’t sufficient data to immediately satisfy the requirement given by minBytes
|
||||
* Default: 5000
|
||||
*/
|
||||
maxWaitTime: HumanDuration | string;
|
||||
retries?: number;
|
||||
};
|
||||
}>;
|
||||
|
||||
/**
|
||||
* (Optional) Timeout for authentication requests.
|
||||
* Default: 10000 ms.
|
||||
*/
|
||||
authenticationTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful connection.
|
||||
* Default: 1000 ms.
|
||||
*/
|
||||
connectionTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful request.
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
requestTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false.
|
||||
* Default: true
|
||||
*/
|
||||
enforceRequestTimeout?: boolean;
|
||||
|
||||
/**
|
||||
* Contains an object per topic for which a Kafka queue
|
||||
* should be used as source of events.
|
||||
*/
|
||||
topics: Array<{
|
||||
/**
|
||||
* (Required) The Backstage topic to publish to
|
||||
*/
|
||||
topic: string;
|
||||
/**
|
||||
* (Required) KafkaConsumer-related configuration.
|
||||
*/
|
||||
kafka: {
|
||||
/**
|
||||
* (Required) The Kafka topics to subscribe to
|
||||
*/
|
||||
topics: string[];
|
||||
/**
|
||||
* (Required) The GroupId to be used by the topic consumers
|
||||
*/
|
||||
groupId: string;
|
||||
|
||||
/**
|
||||
* (Optional) Timeout used to detect failures.
|
||||
* The consumer sends periodic heartbeats to indicate its liveness to the broker.
|
||||
* If no heartbeats are received by the broker before the expiration of this session timeout,
|
||||
* then the broker will remove this consumer from the group and initiate a rebalance
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
sessionTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
|
||||
* Default: 60000 ms.
|
||||
*/
|
||||
rebalanceTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The expected time between heartbeats to the consumer coordinator.
|
||||
* Heartbeats are used to ensure that the consumer's session stays active.
|
||||
* The value must be set lower than session timeout
|
||||
* Default: 3000 ms.
|
||||
*/
|
||||
heartbeatInterval?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The period of time after which we force a refresh of metadata
|
||||
* even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions
|
||||
* Default: 300000 ms (5 minutes).
|
||||
*/
|
||||
metadataMaxAge?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum amount of data per-partition the server will return.
|
||||
* This size must be at least as large as the maximum message size the server allows
|
||||
* or else it is possible for the producer to send messages larger than the consumer can fetch.
|
||||
* If that happens, the consumer can get stuck trying to fetch a large message on a certain partition
|
||||
* Default: 1048576 (1MB)
|
||||
*/
|
||||
maxBytesPerPartition?: number;
|
||||
|
||||
/**
|
||||
* (Optional) Minimum amount of data the server should return for a fetch request, otherwise wait up to maxWaitTime for more data to accumulate.
|
||||
* Default: 1
|
||||
*/
|
||||
minBytes?: number;
|
||||
|
||||
/**
|
||||
* (Optional) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0
|
||||
* Default: 10485760 (10MB)
|
||||
*/
|
||||
maxBytes?: number;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum amount of time the server will block before answering the fetch request
|
||||
* if there isn't sufficient data to immediately satisfy the requirement given by minBytes
|
||||
* Default: 5000
|
||||
*/
|
||||
maxWaitTime?: HumanDuration | string;
|
||||
};
|
||||
}>;
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* Configuration for KafkaPublishingEventConsumer
|
||||
*
|
||||
* Supports multiple named instances as a record where each key is a unique name
|
||||
* for the Kafka producer configuration.
|
||||
*/
|
||||
kafkaPublishingEventConsumer?: {
|
||||
[name: string]: {
|
||||
/**
|
||||
* (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
|
||||
*/
|
||||
clientId: string;
|
||||
/**
|
||||
* (Required) List of brokers in the Kafka cluster to connect to.
|
||||
*/
|
||||
brokers: string[];
|
||||
/**
|
||||
* Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect.
|
||||
* See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options
|
||||
*/
|
||||
ssl?:
|
||||
| {
|
||||
ca?: string[];
|
||||
/** @visibility secret */
|
||||
key?: string;
|
||||
cert?: string;
|
||||
rejectUnauthorized?: boolean;
|
||||
}
|
||||
| boolean;
|
||||
/**
|
||||
* Optional SASL connection parameters.
|
||||
*/
|
||||
sasl?: {
|
||||
mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
|
||||
username: string;
|
||||
/** @visibility secret */
|
||||
password: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Optional retry connection parameters.
|
||||
*/
|
||||
retry?: {
|
||||
/**
|
||||
* (Optional) Maximum wait time for a retry
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
maxRetryTime?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor)
|
||||
* Default: 300 ms.
|
||||
*/
|
||||
initialRetryTime?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Randomization factor
|
||||
* Default: 0.2.
|
||||
*/
|
||||
factor?: number;
|
||||
|
||||
/**
|
||||
* (Optional) Exponential factor
|
||||
* Default: 2.
|
||||
*/
|
||||
multiplier?: number;
|
||||
|
||||
/**
|
||||
* (Optional) Max number of retries per call
|
||||
* Default: 5.
|
||||
*/
|
||||
retries?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* (Optional) Timeout for authentication requests.
|
||||
* Default: 10000 ms.
|
||||
*/
|
||||
authenticationTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful connection.
|
||||
* Default: 1000 ms.
|
||||
*/
|
||||
connectionTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful request.
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
requestTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false.
|
||||
* Default: true
|
||||
*/
|
||||
enforceRequestTimeout?: boolean;
|
||||
|
||||
/**
|
||||
* Contains an object per topic for which a Kafka queue
|
||||
* should be used as destination for events.
|
||||
*/
|
||||
topics: Array<{
|
||||
/**
|
||||
* (Required) The Backstage topic to consume from
|
||||
*/
|
||||
topic: string;
|
||||
/**
|
||||
* (Required) KafkaProducer-related configuration.
|
||||
*/
|
||||
kafka: {
|
||||
/**
|
||||
* (Required) The Kafka topic to publish to
|
||||
*/
|
||||
topic: string;
|
||||
|
||||
/**
|
||||
* (Optional) Allow topic creation when querying metadata for non-existent topics.
|
||||
* Default: true
|
||||
*/
|
||||
allowAutoTopicCreation?: boolean;
|
||||
|
||||
/**
|
||||
* (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 time in ms that the transaction coordinator will wait for a transaction status update
|
||||
* from the producer before proactively aborting the ongoing transaction.
|
||||
* If this value is larger than the `transaction.max.timeout.ms`` setting in the broker, the request will fail with a `InvalidTransactionTimeout` error
|
||||
* Default: 60000 ms.
|
||||
*/
|
||||
transactionTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Experimental. If enabled producer will ensure each message is written exactly once. Acks must be set to -1 ("all").
|
||||
* Retries will default to MAX_SAFE_INTEGER.
|
||||
* Default: false.
|
||||
*/
|
||||
idempotent?: boolean;
|
||||
|
||||
/**
|
||||
* (Optional) Max number of requests that may be in progress at any time. If falsey then no limit.
|
||||
* Default: null.
|
||||
*/
|
||||
maxInFlightRequests?: number;
|
||||
|
||||
/**
|
||||
* 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;
|
||||
};
|
||||
};
|
||||
}>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
|
||||
+163
@@ -0,0 +1,163 @@
|
||||
/*
|
||||
* 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 { ConfigReader } from '@backstage/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;
|
||||
|
||||
jest.mocked(Kafka).mockImplementation(() => mockKafkaClient);
|
||||
|
||||
const mockConfig = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaConsumingEventPublisher: {
|
||||
dev: {
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'backstage-topic',
|
||||
kafka: {
|
||||
topics: ['test-topic'],
|
||||
groupId: 'test-group',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it('should create instances from config', () => {
|
||||
const consumers = KafkaConsumingEventPublisher.fromConfig({
|
||||
config: mockConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
expect(consumers).toBeInstanceOf(Array);
|
||||
expect(consumers).toHaveLength(1);
|
||||
expect(consumers[0]).toBeInstanceOf(KafkaConsumingEventPublisher);
|
||||
});
|
||||
|
||||
it('should return empty array when no config', () => {
|
||||
const consumers = KafkaConsumingEventPublisher.fromConfig({
|
||||
config: new ConfigReader({}),
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
expect(consumers).toEqual([]);
|
||||
});
|
||||
|
||||
it('should start all consumers', async () => {
|
||||
const consumers = KafkaConsumingEventPublisher.fromConfig({
|
||||
config: mockConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
expect(consumers).toHaveLength(1);
|
||||
|
||||
await consumers[0].start();
|
||||
|
||||
expect(mockConsumer.connect).toHaveBeenCalled();
|
||||
expect(mockConsumer.subscribe).toHaveBeenCalledWith({
|
||||
topics: ['test-topic'],
|
||||
});
|
||||
expect(mockConsumer.run).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should shutdown all consumers', async () => {
|
||||
const consumers = KafkaConsumingEventPublisher.fromConfig({
|
||||
config: mockConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
expect(consumers).toHaveLength(1);
|
||||
|
||||
await consumers[0].shutdown();
|
||||
|
||||
expect(mockConsumer.disconnect).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should handle multiple consumer configs', () => {
|
||||
const multiConsumerConfig = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaConsumingEventPublisher: {
|
||||
dev: {
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'topic1',
|
||||
kafka: {
|
||||
topics: ['kafka-topic-1'],
|
||||
groupId: 'group1',
|
||||
},
|
||||
},
|
||||
{
|
||||
topic: 'topic2',
|
||||
kafka: {
|
||||
topics: ['kafka-topic-2'],
|
||||
groupId: 'group2',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const consumers = KafkaConsumingEventPublisher.fromConfig({
|
||||
config: multiConsumerConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
expect(consumers).toHaveLength(1);
|
||||
expect(mockKafkaClient.consumer).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
+109
@@ -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 { EventsService } from '@backstage/plugin-events-node';
|
||||
import { Consumer, Kafka } from 'kafkajs';
|
||||
import {
|
||||
KafkaConsumerConfig,
|
||||
KafkaConsumingEventPublisherConfig,
|
||||
readConsumerConfig,
|
||||
} from './config';
|
||||
import { Config } from '@backstage/config';
|
||||
import { loggerServiceAdapter } from '../utils/LoggerServiceAdapter';
|
||||
import { convertHeadersToMetadata } from '../utils/kafkaTransformers';
|
||||
|
||||
type KafkaConsumer = {
|
||||
consumer: Consumer;
|
||||
config: KafkaConsumerConfig;
|
||||
};
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
export class KafkaConsumingEventPublisher {
|
||||
private readonly kafkaConsumers: KafkaConsumer[];
|
||||
private readonly logger: LoggerService;
|
||||
|
||||
static fromConfig(env: {
|
||||
config: Config;
|
||||
events: EventsService;
|
||||
logger: LoggerService;
|
||||
}): KafkaConsumingEventPublisher[] {
|
||||
const configs = readConsumerConfig(env.config);
|
||||
|
||||
return configs.map(
|
||||
kafkaConfig =>
|
||||
new KafkaConsumingEventPublisher(env.logger, env.events, kafkaConfig),
|
||||
);
|
||||
}
|
||||
|
||||
private constructor(
|
||||
logger: LoggerService,
|
||||
private readonly events: EventsService,
|
||||
config: KafkaConsumingEventPublisherConfig,
|
||||
) {
|
||||
this.logger = logger.child({
|
||||
class: KafkaConsumingEventPublisher.prototype.constructor.name,
|
||||
instance: config.instance,
|
||||
});
|
||||
|
||||
const kafka = new Kafka({
|
||||
...config.kafkaConfig,
|
||||
logCreator: loggerServiceAdapter(this.logger),
|
||||
});
|
||||
|
||||
this.kafkaConsumers = config.kafkaConsumerConfigs.map(consumerConfig => ({
|
||||
consumer: kafka.consumer(consumerConfig.consumerConfig),
|
||||
config: consumerConfig,
|
||||
}));
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
await Promise.all(
|
||||
this.kafkaConsumers.map(async ({ consumer, config }) => {
|
||||
const consumerLogger = this.logger.child({
|
||||
id: `events.kafka.publisher:${config.backstageTopic}`,
|
||||
groupId: config.consumerConfig.groupId,
|
||||
kafkaTopics: config.consumerSubscribeTopics.topics.toString(),
|
||||
backstageTopic: config.backstageTopic,
|
||||
});
|
||||
try {
|
||||
await consumer.connect();
|
||||
await consumer.subscribe(config.consumerSubscribeTopics);
|
||||
|
||||
await consumer.run({
|
||||
eachMessage: async ({ message }) => {
|
||||
this.events.publish({
|
||||
topic: config.backstageTopic,
|
||||
eventPayload: JSON.parse(message.value?.toString()!),
|
||||
metadata: convertHeadersToMetadata(message.headers),
|
||||
});
|
||||
},
|
||||
});
|
||||
} catch (error: any) {
|
||||
consumerLogger.error('Kafka consumer connection failed', error);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
await Promise.all(
|
||||
this.kafkaConsumers.map(({ consumer }) => consumer.disconnect()),
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,275 @@
|
||||
/*
|
||||
* 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 { readConsumerConfig } from './config';
|
||||
|
||||
describe('readConsumerConfig', () => {
|
||||
it('not configured', () => {
|
||||
const publisherConfigs = readConsumerConfig(new ConfigReader({}));
|
||||
|
||||
expect(publisherConfigs).toEqual([]);
|
||||
});
|
||||
|
||||
it('only required fields configured', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaConsumingEventPublisher: {
|
||||
dev: {
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'fake1',
|
||||
kafka: {
|
||||
topics: ['topic-A'],
|
||||
groupId: 'my-group',
|
||||
},
|
||||
},
|
||||
{
|
||||
topic: 'fake2',
|
||||
kafka: {
|
||||
topics: ['topic-B'],
|
||||
groupId: 'my-group',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readConsumerConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const devConfig = publisherConfigs[0];
|
||||
expect(devConfig.instance).toBe('dev');
|
||||
expect(devConfig.kafkaConsumerConfigs.length).toBe(2);
|
||||
|
||||
expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
|
||||
expect(devConfig.kafkaConsumerConfigs).toEqual([
|
||||
{
|
||||
backstageTopic: 'fake1',
|
||||
consumerConfig: {
|
||||
groupId: 'my-group',
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: ['topic-A'],
|
||||
},
|
||||
},
|
||||
{
|
||||
backstageTopic: 'fake2',
|
||||
consumerConfig: {
|
||||
groupId: 'my-group',
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: ['topic-B'],
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('all fields configured', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaConsumingEventPublisher: {
|
||||
dev: {
|
||||
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 = readConsumerConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const devConfig = publisherConfigs[0];
|
||||
expect(devConfig.instance).toBe('dev');
|
||||
|
||||
// Client configuration
|
||||
expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
expect(devConfig.kafkaConfig.ssl).toBeTruthy();
|
||||
expect(devConfig.kafkaConfig.sasl).toStrictEqual({
|
||||
mechanism: 'plain',
|
||||
username: 'username',
|
||||
password: 'password',
|
||||
});
|
||||
expect(devConfig.kafkaConfig.authenticationTimeout).toBe(20000);
|
||||
expect(devConfig.kafkaConfig.connectionTimeout).toBe(1500);
|
||||
expect(devConfig.kafkaConfig.requestTimeout).toBe(20000);
|
||||
expect(devConfig.kafkaConfig.enforceRequestTimeout).toBeFalsy();
|
||||
expect(devConfig.kafkaConfig.retry).toStrictEqual({
|
||||
maxRetryTime: 20000,
|
||||
initialRetryTime: 200,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
|
||||
// Consumer configuration
|
||||
expect(devConfig.kafkaConsumerConfigs.length).toBe(2);
|
||||
|
||||
expect(devConfig.kafkaConsumerConfigs).toEqual([
|
||||
{
|
||||
backstageTopic: 'fake1',
|
||||
consumerConfig: {
|
||||
groupId: 'my-group',
|
||||
sessionTimeout: 20000,
|
||||
rebalanceTimeout: 50000,
|
||||
heartbeatInterval: 2000,
|
||||
metadataMaxAge: 400000,
|
||||
maxBytesPerPartition: 50000,
|
||||
minBytes: 2,
|
||||
maxBytes: 500000,
|
||||
maxWaitTimeInMs: 4000,
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: ['topic-A'],
|
||||
},
|
||||
},
|
||||
{
|
||||
backstageTopic: 'fake2',
|
||||
consumerConfig: {
|
||||
groupId: 'my-group',
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: ['topic-B'],
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('should handle HumanDuration and string values for durations and timeouts', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaConsumingEventPublisher: {
|
||||
dev: {
|
||||
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 = readConsumerConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const devConfig = publisherConfigs[0];
|
||||
expect(devConfig.instance).toBe('dev');
|
||||
|
||||
// Client configuration
|
||||
expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
expect(devConfig.kafkaConfig.authenticationTimeout).toBe(3600000);
|
||||
expect(devConfig.kafkaConfig.connectionTimeout).toBe(86400000);
|
||||
expect(devConfig.kafkaConfig.requestTimeout).toBe(60000);
|
||||
expect(devConfig.kafkaConfig.retry).toStrictEqual({
|
||||
maxRetryTime: 1000,
|
||||
initialRetryTime: 60000,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
|
||||
// Consumer configuration
|
||||
expect(devConfig.kafkaConsumerConfigs.length).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,91 @@
|
||||
/*
|
||||
* 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';
|
||||
import {
|
||||
readKafkaConfig,
|
||||
readOptionalHumanDurationInMs,
|
||||
} from '../utils/config';
|
||||
|
||||
export interface KafkaConsumerConfig {
|
||||
backstageTopic: string;
|
||||
consumerConfig: ConsumerConfig;
|
||||
consumerSubscribeTopics: ConsumerSubscribeTopics;
|
||||
}
|
||||
|
||||
export interface KafkaConsumingEventPublisherConfig {
|
||||
instance: string;
|
||||
kafkaConfig: KafkaConfig;
|
||||
kafkaConsumerConfigs: KafkaConsumerConfig[];
|
||||
}
|
||||
|
||||
const CONFIG_PREFIX_PUBLISHER =
|
||||
'events.modules.kafka.kafkaConsumingEventPublisher';
|
||||
|
||||
export const readConsumerConfig = (
|
||||
config: Config,
|
||||
): KafkaConsumingEventPublisherConfig[] => {
|
||||
const publishers = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER);
|
||||
|
||||
return (
|
||||
publishers?.keys()?.map(publisherKey => {
|
||||
const publisherConfig = publishers.getConfig(publisherKey);
|
||||
|
||||
return {
|
||||
instance: publisherKey,
|
||||
kafkaConfig: readKafkaConfig(publisherConfig),
|
||||
kafkaConsumerConfigs: publisherConfig
|
||||
.getConfigArray('topics')
|
||||
.map(topicConfig => {
|
||||
return {
|
||||
backstageTopic: topicConfig.getString('topic'),
|
||||
consumerConfig: {
|
||||
groupId: topicConfig.getString('kafka.groupId'),
|
||||
sessionTimeout: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.sessionTimeout',
|
||||
),
|
||||
rebalanceTimeout: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.rebalanceTimeout',
|
||||
),
|
||||
heartbeatInterval: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.heartbeatInterval',
|
||||
),
|
||||
metadataMaxAge: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.metadataMaxAge',
|
||||
),
|
||||
maxBytesPerPartition: topicConfig.getOptionalNumber(
|
||||
'kafka.maxBytesPerPartition',
|
||||
),
|
||||
minBytes: topicConfig.getOptionalNumber('kafka.minBytes'),
|
||||
maxBytes: topicConfig.getOptionalNumber('kafka.maxBytes'),
|
||||
maxWaitTimeInMs: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.maxWaitTime',
|
||||
),
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: topicConfig.getStringArray('kafka.topics'),
|
||||
},
|
||||
};
|
||||
}),
|
||||
};
|
||||
}) ?? []
|
||||
);
|
||||
};
|
||||
@@ -0,0 +1,16 @@
|
||||
/*
|
||||
* 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 { eventsModuleKafkaConsumingEventPublisher } from './module';
|
||||
@@ -0,0 +1,122 @@
|
||||
/*
|
||||
* 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 './module';
|
||||
import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher';
|
||||
|
||||
jest.mock('./KafkaConsumingEventPublisher');
|
||||
|
||||
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 mockKafkaConsumingEventPublisher = {
|
||||
start: jest.fn(),
|
||||
shutdown: jest.fn(),
|
||||
} as unknown as KafkaConsumingEventPublisher;
|
||||
|
||||
jest
|
||||
.mocked(KafkaConsumingEventPublisher.fromConfig)
|
||||
.mockReturnValue([mockKafkaConsumingEventPublisher]);
|
||||
|
||||
await startTestBackend({
|
||||
features: [
|
||||
eventsServiceFactory,
|
||||
eventsModuleKafkaConsumingEventPublisher,
|
||||
mockServices.rootConfig.factory({
|
||||
data: {
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaConsumingEventPublisher: {
|
||||
dev: {
|
||||
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(mockKafkaConsumingEventPublisher.start).toHaveBeenCalled();
|
||||
|
||||
// Verify that the shutdown hook was registered
|
||||
expect(mockKafkaConsumingEventPublisher.shutdown).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should handle empty configuration gracefully', async () => {
|
||||
const events = new TestEventsService();
|
||||
const eventsServiceFactory = createServiceFactory({
|
||||
service: eventsServiceRef,
|
||||
deps: {},
|
||||
async factory({}) {
|
||||
return events;
|
||||
},
|
||||
});
|
||||
|
||||
jest.mocked(KafkaConsumingEventPublisher.fromConfig).mockReturnValue([]);
|
||||
|
||||
await startTestBackend({
|
||||
features: [
|
||||
eventsServiceFactory,
|
||||
eventsModuleKafkaConsumingEventPublisher,
|
||||
mockServices.rootConfig.factory({
|
||||
data: {
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
// No kafkaConsumingEventPublisher config
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
});
|
||||
|
||||
// Verify that fromConfig was called but returned empty array
|
||||
expect(KafkaConsumingEventPublisher.fromConfig).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
+10
-10
@@ -17,11 +17,11 @@ import {
|
||||
coreServices,
|
||||
createBackendModule,
|
||||
} from '@backstage/backend-plugin-api';
|
||||
import { KafkaConsumerClient } from '../publisher/KafkaConsumerClient';
|
||||
import { eventsServiceRef } from '@backstage/plugin-events-node';
|
||||
import { KafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher';
|
||||
|
||||
/**
|
||||
* Kafka module for the Events plugin.
|
||||
* Reads messages off of Kafka topics and forwards them into the Backstage events system.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
@@ -34,22 +34,22 @@ export const eventsModuleKafkaConsumingEventPublisher = createBackendModule({
|
||||
config: coreServices.rootConfig,
|
||||
events: eventsServiceRef,
|
||||
logger: coreServices.logger,
|
||||
lifecycle: coreServices.lifecycle,
|
||||
lifecycle: coreServices.rootLifecycle,
|
||||
},
|
||||
async init({ config, logger, events, lifecycle }) {
|
||||
const kafka = KafkaConsumerClient.fromConfig({
|
||||
const consumers = KafkaConsumingEventPublisher.fromConfig({
|
||||
config,
|
||||
events,
|
||||
logger,
|
||||
});
|
||||
|
||||
if (!kafka) {
|
||||
return;
|
||||
}
|
||||
lifecycle.addStartupHook(async () => {
|
||||
await Promise.all(consumers.map(consumer => consumer.start()));
|
||||
});
|
||||
|
||||
await kafka.start();
|
||||
|
||||
lifecycle.addShutdownHook(async () => await kafka.shutdown());
|
||||
lifecycle.addShutdownHook(async () => {
|
||||
await Promise.all(consumers.map(consumer => consumer.shutdown()));
|
||||
});
|
||||
},
|
||||
});
|
||||
},
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
/*
|
||||
* 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 { KafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer';
|
||||
import { Kafka } from 'kafkajs';
|
||||
import { mockServices } from '@backstage/backend-test-utils';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
|
||||
jest.mock('kafkajs');
|
||||
|
||||
describe('KafkaPublishingEventConsumer', () => {
|
||||
const mockLogger = mockServices.logger.mock();
|
||||
const mockEvents = mockServices.events.mock();
|
||||
|
||||
const mockProducer = {
|
||||
connect: jest.fn(),
|
||||
disconnect: jest.fn(),
|
||||
send: jest.fn(),
|
||||
};
|
||||
|
||||
const mockKafkaClient = {
|
||||
producer: jest.fn().mockReturnValue(mockProducer),
|
||||
} as unknown as Kafka;
|
||||
|
||||
jest.mocked(Kafka).mockImplementation(() => mockKafkaClient);
|
||||
|
||||
const mockConfig = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaPublishingEventConsumer: {
|
||||
dev: {
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'backstage-topic',
|
||||
kafka: {
|
||||
topic: 'kafka-topic',
|
||||
allowAutoTopicCreation: true,
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it('should create instances from config', () => {
|
||||
const consumers = KafkaPublishingEventConsumer.fromConfig({
|
||||
config: mockConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
expect(consumers).toHaveLength(1);
|
||||
expect(consumers[0]).toBeInstanceOf(KafkaPublishingEventConsumer);
|
||||
});
|
||||
|
||||
it('should start the consumer and subscribe to events', async () => {
|
||||
const consumers = KafkaPublishingEventConsumer.fromConfig({
|
||||
config: mockConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
await consumers[0].start();
|
||||
|
||||
expect(mockProducer.connect).toHaveBeenCalled();
|
||||
expect(mockEvents.subscribe).toHaveBeenCalledWith({
|
||||
id: 'kafka:publisher:backstage-topic',
|
||||
topics: ['backstage-topic'],
|
||||
onEvent: expect.any(Function),
|
||||
});
|
||||
});
|
||||
|
||||
it('should shutdown the producer', async () => {
|
||||
const consumers = KafkaPublishingEventConsumer.fromConfig({
|
||||
config: mockConfig,
|
||||
events: mockEvents,
|
||||
logger: mockLogger,
|
||||
});
|
||||
|
||||
await consumers[0].shutdown();
|
||||
|
||||
expect(mockProducer.disconnect).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
+115
@@ -0,0 +1,115 @@
|
||||
/*
|
||||
* 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 { Kafka, Producer } from 'kafkajs';
|
||||
import {
|
||||
KafkaPublisherConfig,
|
||||
KafkaPublishingEventConsumerConfig,
|
||||
readPublisherConfig,
|
||||
} from './config';
|
||||
import { Config } from '@backstage/config';
|
||||
import { loggerServiceAdapter } from '../utils/LoggerServiceAdapter';
|
||||
import { payloadToBuffer } from '../utils/kafkaTransformers';
|
||||
|
||||
type KafkaPublisher = {
|
||||
producer: Producer;
|
||||
config: KafkaPublisherConfig;
|
||||
};
|
||||
|
||||
/**
|
||||
* This class subscribes to Backstage internal events and publishes them to Kafka topics.
|
||||
* The internal event payload will be serialized and sent to the configured Kafka topic.
|
||||
*/
|
||||
export class KafkaPublishingEventConsumer {
|
||||
private readonly kafkaPublishers: KafkaPublisher[];
|
||||
private readonly logger: LoggerService;
|
||||
|
||||
static fromConfig(env: {
|
||||
config: Config;
|
||||
events: EventsService;
|
||||
logger: LoggerService;
|
||||
}): KafkaPublishingEventConsumer[] {
|
||||
const configs = readPublisherConfig(env.config);
|
||||
|
||||
return configs.map(
|
||||
kafkaConfig =>
|
||||
new KafkaPublishingEventConsumer(env.logger, env.events, kafkaConfig),
|
||||
);
|
||||
}
|
||||
|
||||
private constructor(
|
||||
logger: LoggerService,
|
||||
private readonly events: EventsService,
|
||||
config: KafkaPublishingEventConsumerConfig,
|
||||
) {
|
||||
this.logger = logger.child({
|
||||
class: KafkaPublishingEventConsumer.prototype.constructor.name,
|
||||
instance: config.instance,
|
||||
});
|
||||
|
||||
const kafka = new Kafka({
|
||||
...config.kafkaConfig,
|
||||
logCreator: loggerServiceAdapter(this.logger),
|
||||
});
|
||||
|
||||
this.kafkaPublishers = config.kafkaPublisherConfigs.map(
|
||||
publisherConfig => ({
|
||||
producer: kafka.producer(publisherConfig.producerConfig),
|
||||
config: publisherConfig,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
await Promise.all(
|
||||
this.kafkaPublishers.map(async ({ producer, config }) => {
|
||||
try {
|
||||
await producer.connect();
|
||||
|
||||
this.events.subscribe({
|
||||
id: `kafka:publisher:${config.backstageTopic}`,
|
||||
topics: [config.backstageTopic],
|
||||
onEvent: async (params: EventParams) => {
|
||||
await producer.send({
|
||||
topic: config.kafkaTopic,
|
||||
messages: [
|
||||
{
|
||||
value: payloadToBuffer(params.eventPayload),
|
||||
},
|
||||
],
|
||||
});
|
||||
},
|
||||
});
|
||||
this.logger.info(
|
||||
`Subscribed to EventService, publishing events to external topic: ${config.backstageTopic}`,
|
||||
);
|
||||
} catch (error: any) {
|
||||
this.logger.error(
|
||||
`Kafka producer connection failed for topic ${config.backstageTopic}`,
|
||||
error,
|
||||
);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
await Promise.all(
|
||||
this.kafkaPublishers.map(({ producer }) => producer.disconnect()),
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,360 @@
|
||||
/*
|
||||
* 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 { readPublisherConfig } from './config';
|
||||
|
||||
describe('readPublisherConfig', () => {
|
||||
it('not configured', () => {
|
||||
const publisherConfigs = readPublisherConfig(new ConfigReader({}));
|
||||
|
||||
expect(publisherConfigs).toEqual([]);
|
||||
});
|
||||
|
||||
it('only required fields configured', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaPublishingEventConsumer: {
|
||||
dev: {
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'fake1',
|
||||
kafka: {
|
||||
topic: 'topic-A',
|
||||
},
|
||||
},
|
||||
{
|
||||
topic: 'fake2',
|
||||
kafka: {
|
||||
topic: 'topic-B',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readPublisherConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const devConfig = publisherConfigs[0];
|
||||
expect(devConfig.instance).toBe('dev');
|
||||
expect(devConfig.kafkaPublisherConfigs.length).toBe(2);
|
||||
|
||||
expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
|
||||
expect(devConfig.kafkaPublisherConfigs).toEqual([
|
||||
{
|
||||
backstageTopic: 'fake1',
|
||||
kafkaTopic: 'topic-A',
|
||||
producerConfig: {
|
||||
allowAutoTopicCreation: undefined,
|
||||
metadataMaxAge: undefined,
|
||||
transactionTimeout: undefined,
|
||||
idempotent: undefined,
|
||||
maxInFlightRequests: undefined,
|
||||
retry: {},
|
||||
},
|
||||
},
|
||||
{
|
||||
backstageTopic: 'fake2',
|
||||
kafkaTopic: 'topic-B',
|
||||
producerConfig: {
|
||||
allowAutoTopicCreation: undefined,
|
||||
metadataMaxAge: undefined,
|
||||
transactionTimeout: undefined,
|
||||
idempotent: undefined,
|
||||
maxInFlightRequests: undefined,
|
||||
retry: {},
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('all fields configured', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaPublishingEventConsumer: {
|
||||
dev: {
|
||||
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: {
|
||||
topic: 'topic-A',
|
||||
allowAutoTopicCreation: true,
|
||||
metadataMaxAge: { milliseconds: 400000 },
|
||||
transactionTimeout: { milliseconds: 30000 },
|
||||
idempotent: true,
|
||||
maxInFlightRequests: 5,
|
||||
retry: {
|
||||
maxRetryTime: { milliseconds: 15000 },
|
||||
initialRetryTime: { milliseconds: 100 },
|
||||
factor: '0.2',
|
||||
multiplier: '2',
|
||||
retries: '5',
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
topic: 'fake2',
|
||||
kafka: {
|
||||
topic: 'topic-B',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readPublisherConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const devConfig = publisherConfigs[0];
|
||||
expect(devConfig.instance).toBe('dev');
|
||||
|
||||
// Client configuration
|
||||
expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
expect(devConfig.kafkaConfig.ssl).toBeTruthy();
|
||||
expect(devConfig.kafkaConfig.sasl).toStrictEqual({
|
||||
mechanism: 'plain',
|
||||
username: 'username',
|
||||
password: 'password',
|
||||
});
|
||||
expect(devConfig.kafkaConfig.authenticationTimeout).toBe(20000);
|
||||
expect(devConfig.kafkaConfig.connectionTimeout).toBe(1500);
|
||||
expect(devConfig.kafkaConfig.requestTimeout).toBe(20000);
|
||||
expect(devConfig.kafkaConfig.enforceRequestTimeout).toBeFalsy();
|
||||
expect(devConfig.kafkaConfig.retry).toStrictEqual({
|
||||
maxRetryTime: 20000,
|
||||
initialRetryTime: 200,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
|
||||
// Publisher configuration
|
||||
expect(devConfig.kafkaPublisherConfigs.length).toBe(2);
|
||||
|
||||
expect(devConfig.kafkaPublisherConfigs).toEqual([
|
||||
{
|
||||
backstageTopic: 'fake1',
|
||||
kafkaTopic: 'topic-A',
|
||||
producerConfig: {
|
||||
allowAutoTopicCreation: true,
|
||||
metadataMaxAge: 400000,
|
||||
transactionTimeout: 30000,
|
||||
idempotent: true,
|
||||
maxInFlightRequests: 5,
|
||||
retry: {
|
||||
maxRetryTime: 15000,
|
||||
initialRetryTime: 100,
|
||||
factor: 0.2,
|
||||
multiplier: 2,
|
||||
retries: 5,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
backstageTopic: 'fake2',
|
||||
kafkaTopic: 'topic-B',
|
||||
producerConfig: {
|
||||
allowAutoTopicCreation: undefined,
|
||||
metadataMaxAge: undefined,
|
||||
transactionTimeout: undefined,
|
||||
idempotent: undefined,
|
||||
maxInFlightRequests: undefined,
|
||||
retry: {},
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('should handle HumanDuration and string values for durations and timeouts', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaPublishingEventConsumer: {
|
||||
dev: {
|
||||
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: [
|
||||
{
|
||||
topic: 'fake1',
|
||||
kafka: {
|
||||
topic: 'topic-A',
|
||||
metadataMaxAge: { seconds: 300 },
|
||||
transactionTimeout: '30s',
|
||||
},
|
||||
},
|
||||
],
|
||||
requestTimeout: '1m',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readPublisherConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const devConfig = publisherConfigs[0];
|
||||
expect(devConfig.instance).toBe('dev');
|
||||
|
||||
// Client configuration
|
||||
expect(devConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
expect(devConfig.kafkaConfig.authenticationTimeout).toBe(3600000);
|
||||
expect(devConfig.kafkaConfig.connectionTimeout).toBe(86400000);
|
||||
expect(devConfig.kafkaConfig.requestTimeout).toBe(60000);
|
||||
expect(devConfig.kafkaConfig.retry).toStrictEqual({
|
||||
maxRetryTime: 1000,
|
||||
initialRetryTime: 60000,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
|
||||
// Publisher configuration
|
||||
expect(devConfig.kafkaPublisherConfigs.length).toBe(1);
|
||||
expect(
|
||||
devConfig.kafkaPublisherConfigs[0].producerConfig.metadataMaxAge,
|
||||
).toBe(300000);
|
||||
expect(
|
||||
devConfig.kafkaPublisherConfigs[0].producerConfig.transactionTimeout,
|
||||
).toBe(30000);
|
||||
});
|
||||
|
||||
it('should handle multiple instances', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaPublishingEventConsumer: {
|
||||
dev: {
|
||||
clientId: 'backstage-dev',
|
||||
brokers: ['kafka-dev:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'dev-topic',
|
||||
kafka: {
|
||||
topic: 'kafka-dev-topic',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
prod: {
|
||||
clientId: 'backstage-prod',
|
||||
brokers: ['kafka-prod1:9092', 'kafka-prod2:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'prod-topic',
|
||||
kafka: {
|
||||
topic: 'kafka-prod-topic',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readPublisherConfig(config);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(2);
|
||||
|
||||
const devConfig = publisherConfigs.find(c => c.instance === 'dev')!;
|
||||
expect(devConfig.kafkaConfig.clientId).toBe('backstage-dev');
|
||||
expect(devConfig.kafkaConfig.brokers).toEqual(['kafka-dev:9092']);
|
||||
expect(devConfig.kafkaPublisherConfigs).toHaveLength(1);
|
||||
expect(devConfig.kafkaPublisherConfigs[0].backstageTopic).toBe('dev-topic');
|
||||
|
||||
const prodConfig = publisherConfigs.find(c => c.instance === 'prod')!;
|
||||
expect(prodConfig.kafkaConfig.clientId).toBe('backstage-prod');
|
||||
expect(prodConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka-prod1:9092',
|
||||
'kafka-prod2:9092',
|
||||
]);
|
||||
expect(prodConfig.kafkaPublisherConfigs).toHaveLength(1);
|
||||
expect(prodConfig.kafkaPublisherConfigs[0].backstageTopic).toBe(
|
||||
'prod-topic',
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,82 @@
|
||||
/*
|
||||
* 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 {
|
||||
readKafkaConfig,
|
||||
readOptionalHumanDurationInMs,
|
||||
readRetryConfig,
|
||||
} from '../utils/config';
|
||||
import { KafkaConfig, ProducerConfig } from 'kafkajs';
|
||||
|
||||
export interface KafkaPublisherConfig {
|
||||
backstageTopic: string;
|
||||
kafkaTopic: string;
|
||||
producerConfig: ProducerConfig;
|
||||
}
|
||||
|
||||
export interface KafkaPublishingEventConsumerConfig {
|
||||
instance: string;
|
||||
kafkaConfig: KafkaConfig;
|
||||
kafkaPublisherConfigs: KafkaPublisherConfig[];
|
||||
}
|
||||
|
||||
const CONFIG_PREFIX_PUBLISHER =
|
||||
'events.modules.kafka.kafkaPublishingEventConsumer';
|
||||
|
||||
export const readPublisherConfig = (
|
||||
config: Config,
|
||||
): KafkaPublishingEventConsumerConfig[] => {
|
||||
const publishers = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER);
|
||||
|
||||
return (
|
||||
publishers?.keys()?.map(publisherKey => {
|
||||
const publisherConfig = publishers.getConfig(publisherKey);
|
||||
|
||||
return {
|
||||
instance: publisherKey,
|
||||
kafkaConfig: readKafkaConfig(publisherConfig),
|
||||
kafkaPublisherConfigs: publisherConfig
|
||||
.getConfigArray('topics')
|
||||
.map(topicConfig => {
|
||||
return {
|
||||
backstageTopic: topicConfig.getString('topic'),
|
||||
kafkaTopic: topicConfig.getString('kafka.topic'),
|
||||
producerConfig: {
|
||||
allowAutoTopicCreation: topicConfig.getOptionalBoolean(
|
||||
'kafka.allowAutoTopicCreation',
|
||||
),
|
||||
metadataMaxAge: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.metadataMaxAge',
|
||||
),
|
||||
transactionTimeout: readOptionalHumanDurationInMs(
|
||||
topicConfig,
|
||||
'kafka.transactionTimeout',
|
||||
),
|
||||
idempotent: topicConfig.getOptionalBoolean('kafka.idempotent'),
|
||||
maxInFlightRequests: topicConfig.getOptionalNumber(
|
||||
'kafka.maxInFlightRequests',
|
||||
),
|
||||
retry: readRetryConfig(
|
||||
topicConfig.getOptionalConfig('kafka.retry'),
|
||||
),
|
||||
},
|
||||
};
|
||||
}),
|
||||
};
|
||||
}) ?? []
|
||||
);
|
||||
};
|
||||
@@ -0,0 +1,16 @@
|
||||
/*
|
||||
* 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 { eventsModuleKafkaPublishingEventConsumer } from './module';
|
||||
@@ -0,0 +1,122 @@
|
||||
/*
|
||||
* 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 { eventsModuleKafkaPublishingEventConsumer } from './module';
|
||||
import { KafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer';
|
||||
|
||||
jest.mock('./KafkaPublishingEventConsumer');
|
||||
|
||||
describe('eventsModuleKafkaPublishingEventConsumer', () => {
|
||||
it('should be correctly wired and set up', async () => {
|
||||
const events = new TestEventsService();
|
||||
const eventsServiceFactory = createServiceFactory({
|
||||
service: eventsServiceRef,
|
||||
deps: {},
|
||||
async factory({}) {
|
||||
return events;
|
||||
},
|
||||
});
|
||||
|
||||
const mockKafkaPublishingEventConsumer = {
|
||||
start: jest.fn(),
|
||||
shutdown: jest.fn(),
|
||||
} as unknown as KafkaPublishingEventConsumer;
|
||||
|
||||
jest
|
||||
.mocked(KafkaPublishingEventConsumer.fromConfig)
|
||||
.mockReturnValue([mockKafkaPublishingEventConsumer]);
|
||||
|
||||
await startTestBackend({
|
||||
features: [
|
||||
eventsServiceFactory,
|
||||
eventsModuleKafkaPublishingEventConsumer,
|
||||
mockServices.rootConfig.factory({
|
||||
data: {
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
kafkaPublishingEventConsumer: {
|
||||
dev: {
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
topics: [
|
||||
{
|
||||
topic: 'fake1',
|
||||
kafka: {
|
||||
topic: 'topic-A',
|
||||
},
|
||||
},
|
||||
{
|
||||
topic: 'fake2',
|
||||
kafka: {
|
||||
topic: 'topic-B',
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
});
|
||||
|
||||
// Verify that the Kafka publishing event consumer was started
|
||||
expect(mockKafkaPublishingEventConsumer.start).toHaveBeenCalled();
|
||||
|
||||
// Verify that the shutdown hook was registered (but not called yet)
|
||||
expect(mockKafkaPublishingEventConsumer.shutdown).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should handle empty configuration gracefully', async () => {
|
||||
const events = new TestEventsService();
|
||||
const eventsServiceFactory = createServiceFactory({
|
||||
service: eventsServiceRef,
|
||||
deps: {},
|
||||
async factory({}) {
|
||||
return events;
|
||||
},
|
||||
});
|
||||
|
||||
jest.mocked(KafkaPublishingEventConsumer.fromConfig).mockReturnValue([]);
|
||||
|
||||
await startTestBackend({
|
||||
features: [
|
||||
eventsServiceFactory,
|
||||
eventsModuleKafkaPublishingEventConsumer,
|
||||
mockServices.rootConfig.factory({
|
||||
data: {
|
||||
events: {
|
||||
modules: {
|
||||
kafka: {
|
||||
// No kafkaPublishingEventConsumer config
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
});
|
||||
|
||||
// Verify that fromConfig was called but returned empty array
|
||||
expect(KafkaPublishingEventConsumer.fromConfig).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -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 { eventsServiceRef } from '@backstage/plugin-events-node';
|
||||
import { KafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer';
|
||||
|
||||
/**
|
||||
* Reads internal Backstage events and forwards them to Kafka topics.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export const eventsModuleKafkaPublishingEventConsumer = createBackendModule({
|
||||
pluginId: 'events',
|
||||
moduleId: 'kafka-publishing-event-consumer',
|
||||
register(env) {
|
||||
env.registerInit({
|
||||
deps: {
|
||||
config: coreServices.rootConfig,
|
||||
events: eventsServiceRef,
|
||||
logger: coreServices.logger,
|
||||
lifecycle: coreServices.rootLifecycle,
|
||||
},
|
||||
async init({ config, logger, events, lifecycle }) {
|
||||
const consumers = KafkaPublishingEventConsumer.fromConfig({
|
||||
config,
|
||||
events,
|
||||
logger,
|
||||
});
|
||||
|
||||
lifecycle.addStartupHook(async () => {
|
||||
await Promise.all(consumers.map(consumer => consumer.start()));
|
||||
});
|
||||
|
||||
lifecycle.addShutdownHook(async () => {
|
||||
await Promise.all(consumers.map(consumer => consumer.shutdown()));
|
||||
});
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -13,14 +13,22 @@
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
import { createBackendFeatureLoader } from '@backstage/backend-plugin-api';
|
||||
import { eventsModuleKafkaConsumingEventPublisher } from './KafkaConsumingEventPublisher';
|
||||
import { eventsModuleKafkaPublishingEventConsumer } from './KafkaPublishingEventConsumer';
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* adding Kafka-based event handling:
|
||||
* - Consumer: receives events from Kafka topics and passes them to the internal event broker
|
||||
* - Publisher: receives internal events and publishes them to Kafka topics
|
||||
*
|
||||
* @packageDocumentation
|
||||
*/
|
||||
|
||||
export { eventsModuleKafkaConsumingEventPublisher as default } from './service/eventsModuleKafkaConsumingEventPublisher';
|
||||
export default createBackendFeatureLoader({
|
||||
*loader() {
|
||||
yield eventsModuleKafkaConsumingEventPublisher;
|
||||
yield eventsModuleKafkaPublishingEventConsumer;
|
||||
},
|
||||
});
|
||||
|
||||
@@ -1,132 +0,0 @@
|
||||
/*
|
||||
* 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();
|
||||
});
|
||||
});
|
||||
@@ -1,77 +0,0 @@
|
||||
/*
|
||||
* 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
|
||||
*/
|
||||
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
@@ -1,92 +0,0 @@
|
||||
/*
|
||||
* 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();
|
||||
});
|
||||
});
|
||||
@@ -1,109 +0,0 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
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 readonly events: EventsService;
|
||||
|
||||
private constructor(
|
||||
kafkaClient: Kafka,
|
||||
logger: LoggerService,
|
||||
events: EventsService,
|
||||
config: KafkaConsumerConfig,
|
||||
) {
|
||||
this.events = events;
|
||||
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;
|
||||
};
|
||||
}
|
||||
@@ -1,250 +0,0 @@
|
||||
/*
|
||||
* 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);
|
||||
});
|
||||
});
|
||||
@@ -1,153 +0,0 @@
|
||||
/*
|
||||
* 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 { durationToMilliseconds } from '@backstage/types';
|
||||
import { ConsumerConfig, ConsumerSubscribeTopics, KafkaConfig } from 'kafkajs';
|
||||
|
||||
export interface KafkaConsumerConfig {
|
||||
backstageTopic: string;
|
||||
consumerConfig: ConsumerConfig;
|
||||
consumerSubscribeTopics: ConsumerSubscribeTopics;
|
||||
}
|
||||
|
||||
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 durationToMilliseconds(humanDuration);
|
||||
};
|
||||
|
||||
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
@@ -1,85 +0,0 @@
|
||||
/*
|
||||
* 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();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,235 @@
|
||||
/*
|
||||
* 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 { readRetryConfig, readKafkaConfig } from './config';
|
||||
|
||||
describe('readRetryConfig', () => {
|
||||
it('should return empty object when config is undefined', () => {
|
||||
const result = readRetryConfig(undefined);
|
||||
expect(result).toEqual({});
|
||||
});
|
||||
|
||||
it('should return empty object when config is empty', () => {
|
||||
const config = new ConfigReader({});
|
||||
const result = readRetryConfig(config);
|
||||
expect(result).toEqual({});
|
||||
});
|
||||
|
||||
it('should read retry configuration with milliseconds', () => {
|
||||
const config = new ConfigReader({
|
||||
maxRetryTime: { milliseconds: 20000 },
|
||||
initialRetryTime: { milliseconds: 200 },
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
|
||||
const result = readRetryConfig(config);
|
||||
|
||||
expect(result).toEqual({
|
||||
maxRetryTime: 20000,
|
||||
initialRetryTime: 200,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
});
|
||||
|
||||
it('should read retry configuration with string values', () => {
|
||||
const config = new ConfigReader({
|
||||
maxRetryTime: { seconds: 20 },
|
||||
initialRetryTime: { minutes: 1 },
|
||||
factor: '0.4',
|
||||
multiplier: '4',
|
||||
retries: '10',
|
||||
});
|
||||
|
||||
const result = readRetryConfig(config);
|
||||
|
||||
expect(result).toEqual({
|
||||
maxRetryTime: 20000,
|
||||
initialRetryTime: 60000,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
});
|
||||
});
|
||||
|
||||
it('should handle HumanDuration values', () => {
|
||||
const config = new ConfigReader({
|
||||
maxRetryTime: { hours: 1 },
|
||||
initialRetryTime: { days: 1 },
|
||||
});
|
||||
|
||||
const result = readRetryConfig(config);
|
||||
|
||||
expect(result).toEqual({
|
||||
maxRetryTime: 3600000,
|
||||
initialRetryTime: 86400000,
|
||||
factor: undefined,
|
||||
multiplier: undefined,
|
||||
retries: undefined,
|
||||
});
|
||||
});
|
||||
|
||||
it('should handle partial configuration', () => {
|
||||
const config = new ConfigReader({
|
||||
maxRetryTime: { milliseconds: 15000 },
|
||||
retries: 5,
|
||||
});
|
||||
|
||||
const result = readRetryConfig(config);
|
||||
|
||||
expect(result).toEqual({
|
||||
maxRetryTime: 15000,
|
||||
initialRetryTime: undefined,
|
||||
factor: undefined,
|
||||
multiplier: undefined,
|
||||
retries: 5,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('readKafkaConfig', () => {
|
||||
it('should read minimal kafka configuration', () => {
|
||||
const config = new ConfigReader({
|
||||
clientId: 'test-client',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
});
|
||||
|
||||
const result = readKafkaConfig(config);
|
||||
|
||||
expect(result).toEqual({
|
||||
clientId: 'test-client',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
authenticationTimeout: undefined,
|
||||
connectionTimeout: undefined,
|
||||
requestTimeout: undefined,
|
||||
enforceRequestTimeout: undefined,
|
||||
ssl: undefined,
|
||||
sasl: undefined,
|
||||
retry: {},
|
||||
});
|
||||
});
|
||||
|
||||
it('should read full kafka configuration with all optional fields', () => {
|
||||
const config = new ConfigReader({
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
authenticationTimeout: { milliseconds: 20000 },
|
||||
connectionTimeout: { milliseconds: 1500 },
|
||||
requestTimeout: { milliseconds: 20000 },
|
||||
enforceRequestTimeout: false,
|
||||
ssl: true,
|
||||
sasl: {
|
||||
mechanism: 'plain',
|
||||
username: 'username',
|
||||
password: 'password',
|
||||
},
|
||||
retry: {
|
||||
maxRetryTime: { milliseconds: 20000 },
|
||||
initialRetryTime: { milliseconds: 200 },
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
},
|
||||
});
|
||||
|
||||
const result = readKafkaConfig(config);
|
||||
|
||||
expect(result).toEqual({
|
||||
clientId: 'backstage-events',
|
||||
brokers: ['kafka1:9092', 'kafka2:9092'],
|
||||
authenticationTimeout: 20000,
|
||||
connectionTimeout: 1500,
|
||||
requestTimeout: 20000,
|
||||
enforceRequestTimeout: false,
|
||||
ssl: true,
|
||||
sasl: {
|
||||
mechanism: 'plain',
|
||||
username: 'username',
|
||||
password: 'password',
|
||||
},
|
||||
retry: {
|
||||
maxRetryTime: 20000,
|
||||
initialRetryTime: 200,
|
||||
factor: 0.4,
|
||||
multiplier: 4,
|
||||
retries: 10,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('should handle HumanDuration values for timeouts', () => {
|
||||
const config = new ConfigReader({
|
||||
clientId: 'test-client',
|
||||
brokers: ['kafka:9092'],
|
||||
authenticationTimeout: { hours: 1 },
|
||||
connectionTimeout: { days: 1 },
|
||||
requestTimeout: { minutes: 5 },
|
||||
});
|
||||
|
||||
const result = readKafkaConfig(config);
|
||||
|
||||
expect(result.authenticationTimeout).toBe(3600000);
|
||||
expect(result.connectionTimeout).toBe(86400000);
|
||||
expect(result.requestTimeout).toBe(300000);
|
||||
});
|
||||
|
||||
it('should handle complex SSL configuration', () => {
|
||||
const config = new ConfigReader({
|
||||
clientId: 'secure-client',
|
||||
brokers: ['secure-kafka:9093'],
|
||||
ssl: {
|
||||
rejectUnauthorized: false,
|
||||
ca: 'ca-certificate',
|
||||
key: 'client-key',
|
||||
cert: 'client-cert',
|
||||
},
|
||||
});
|
||||
|
||||
const result = readKafkaConfig(config);
|
||||
|
||||
expect(result.ssl).toEqual({
|
||||
rejectUnauthorized: false,
|
||||
ca: 'ca-certificate',
|
||||
key: 'client-key',
|
||||
cert: 'client-cert',
|
||||
});
|
||||
});
|
||||
|
||||
it('should handle complex SASL configuration', () => {
|
||||
const config = new ConfigReader({
|
||||
clientId: 'sasl-client',
|
||||
brokers: ['sasl-kafka:9094'],
|
||||
sasl: {
|
||||
mechanism: 'scram-sha-256',
|
||||
username: 'kafka-user',
|
||||
password: 'kafka-password',
|
||||
authorizationIdentity: 'authz-user',
|
||||
},
|
||||
});
|
||||
|
||||
const result = readKafkaConfig(config);
|
||||
|
||||
expect(result.sasl).toEqual({
|
||||
mechanism: 'scram-sha-256',
|
||||
username: 'kafka-user',
|
||||
password: 'kafka-password',
|
||||
authorizationIdentity: 'authz-user',
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,84 @@
|
||||
/*
|
||||
* 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 { durationToMilliseconds } from '@backstage/types';
|
||||
import { KafkaConfig, RetryOptions } from 'kafkajs';
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
export const readOptionalHumanDurationInMs = (
|
||||
config: Config,
|
||||
key: string,
|
||||
): number | undefined => {
|
||||
const humanDuration = config.has(key)
|
||||
? readDurationFromConfig(config, { key })
|
||||
: undefined;
|
||||
|
||||
if (!humanDuration) return undefined;
|
||||
|
||||
return durationToMilliseconds(humanDuration);
|
||||
};
|
||||
|
||||
/**
|
||||
* Reads retry configuration options from the provided config object.
|
||||
*
|
||||
* @param config - The configuration object to read retry options from, or undefined.
|
||||
* @returns A RetryOptions object with optional retry settings, or an empty object if config is undefined.
|
||||
*/
|
||||
export const readRetryConfig = (config: Config | undefined): RetryOptions => {
|
||||
if (!config) {
|
||||
return {};
|
||||
}
|
||||
|
||||
return {
|
||||
maxRetryTime: readOptionalHumanDurationInMs(config, 'maxRetryTime'),
|
||||
initialRetryTime: readOptionalHumanDurationInMs(config, 'initialRetryTime'),
|
||||
factor: config.getOptionalNumber('factor'),
|
||||
multiplier: config.getOptionalNumber('multiplier'),
|
||||
retries: config.getOptionalNumber('retries'),
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* Reads Kafka configuration from the provided config object.
|
||||
*
|
||||
* @param config - The configuration object containing Kafka settings.
|
||||
* @returns A KafkaConfig object with all necessary Kafka connection and authentication settings.
|
||||
*/
|
||||
export const readKafkaConfig = (config: Config): KafkaConfig => {
|
||||
return {
|
||||
clientId: config.getString('clientId'),
|
||||
brokers: config.getStringArray('brokers'),
|
||||
authenticationTimeout: readOptionalHumanDurationInMs(
|
||||
config,
|
||||
'authenticationTimeout',
|
||||
),
|
||||
connectionTimeout: readOptionalHumanDurationInMs(
|
||||
config,
|
||||
'connectionTimeout',
|
||||
),
|
||||
requestTimeout: readOptionalHumanDurationInMs(config, 'requestTimeout'),
|
||||
enforceRequestTimeout: config.getOptionalBoolean('enforceRequestTimeout'),
|
||||
ssl: config.getOptional('ssl') as KafkaConfig['ssl'],
|
||||
sasl: config.getOptional('sasl') as KafkaConfig['sasl'],
|
||||
retry: readRetryConfig(config.getOptionalConfig('retry')),
|
||||
};
|
||||
};
|
||||
@@ -0,0 +1,137 @@
|
||||
/*
|
||||
* 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 { convertHeadersToMetadata, payloadToBuffer } from './kafkaTransformers';
|
||||
|
||||
describe('kafka-transformers', () => {
|
||||
describe('convertHeadersToMetadata', () => {
|
||||
it('should return undefined when headers is undefined', () => {
|
||||
const result = convertHeadersToMetadata(undefined);
|
||||
expect(result).toBeUndefined();
|
||||
});
|
||||
|
||||
it('should convert string headers to metadata', () => {
|
||||
const headers = {
|
||||
'content-type': 'application/json',
|
||||
'user-id': '12345',
|
||||
};
|
||||
|
||||
const result = convertHeadersToMetadata(headers);
|
||||
|
||||
expect(result).toEqual({
|
||||
'content-type': 'application/json',
|
||||
'user-id': '12345',
|
||||
});
|
||||
});
|
||||
|
||||
it('should convert Buffer headers to string metadata', () => {
|
||||
const headers = {
|
||||
'content-type': Buffer.from('application/json'),
|
||||
'correlation-id': Buffer.from('abc-123'),
|
||||
};
|
||||
|
||||
const result = convertHeadersToMetadata(headers);
|
||||
|
||||
expect(result).toEqual({
|
||||
'content-type': 'application/json',
|
||||
'correlation-id': 'abc-123',
|
||||
});
|
||||
});
|
||||
|
||||
it('should convert array headers to string array metadata', () => {
|
||||
const headers = {
|
||||
tags: ['tag1', 'tag2'],
|
||||
'buffer-tags': [Buffer.from('tag3'), Buffer.from('tag4')],
|
||||
'mixed-tags': ['tag5', Buffer.from('tag6')],
|
||||
};
|
||||
|
||||
const result = convertHeadersToMetadata(headers);
|
||||
|
||||
expect(result).toEqual({
|
||||
tags: ['tag1', 'tag2'],
|
||||
'buffer-tags': ['tag3', 'tag4'],
|
||||
'mixed-tags': ['tag5', 'tag6'],
|
||||
});
|
||||
});
|
||||
|
||||
it('should handle mixed header types', () => {
|
||||
const headers = {
|
||||
'string-header': 'value',
|
||||
'buffer-header': Buffer.from('buffer-value'),
|
||||
'array-header': ['item1', Buffer.from('item2')],
|
||||
'undefined-header': undefined,
|
||||
};
|
||||
|
||||
const result = convertHeadersToMetadata(headers);
|
||||
|
||||
expect(result).toEqual({
|
||||
'string-header': 'value',
|
||||
'buffer-header': 'buffer-value',
|
||||
'array-header': ['item1', 'item2'],
|
||||
'undefined-header': undefined,
|
||||
});
|
||||
});
|
||||
|
||||
it('should handle empty headers object', () => {
|
||||
const headers = {};
|
||||
const result = convertHeadersToMetadata(headers);
|
||||
expect(result).toEqual({});
|
||||
});
|
||||
});
|
||||
|
||||
describe('payloadToBuffer', () => {
|
||||
it('should return the same Buffer when payload is already a Buffer', () => {
|
||||
const originalBuffer = Buffer.from('test data');
|
||||
|
||||
const result = payloadToBuffer(originalBuffer);
|
||||
|
||||
expect(result).toBe(originalBuffer);
|
||||
expect(Buffer.isBuffer(result)).toBe(true);
|
||||
});
|
||||
|
||||
it('should convert string to Buffer', () => {
|
||||
const payload = 'hello world';
|
||||
|
||||
const result = payloadToBuffer(payload);
|
||||
|
||||
expect(Buffer.isBuffer(result)).toBe(true);
|
||||
expect(result.toString()).toBe('hello world');
|
||||
});
|
||||
|
||||
it('should convert object to JSON Buffer', () => {
|
||||
const payload = { name: 'John', age: 30 };
|
||||
|
||||
const result = payloadToBuffer(payload);
|
||||
|
||||
expect(Buffer.isBuffer(result)).toBe(true);
|
||||
expect(JSON.parse(result.toString())).toEqual(payload);
|
||||
});
|
||||
|
||||
it('should convert array to JSON Buffer', () => {
|
||||
const payload = [1, 2, 3, 'test'];
|
||||
|
||||
const result = payloadToBuffer(payload);
|
||||
|
||||
expect(Buffer.isBuffer(result)).toBe(true);
|
||||
expect(JSON.parse(result.toString())).toEqual(payload);
|
||||
});
|
||||
|
||||
it('should convert primitives to JSON Buffer', () => {
|
||||
expect(payloadToBuffer(42).toString()).toBe('42');
|
||||
expect(payloadToBuffer(true).toString()).toBe('true');
|
||||
expect(payloadToBuffer(null).toString()).toBe('null');
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,49 @@
|
||||
/*
|
||||
* 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 { EventParams } from '@backstage/plugin-events-node';
|
||||
import { IHeaders } from 'kafkajs';
|
||||
|
||||
type EventMetadata = EventParams['metadata'];
|
||||
|
||||
export const convertHeadersToMetadata = (
|
||||
headers: IHeaders | undefined,
|
||||
): EventMetadata => {
|
||||
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;
|
||||
};
|
||||
|
||||
export const payloadToBuffer = (payload: unknown): Buffer => {
|
||||
if (Buffer.isBuffer(payload)) {
|
||||
return payload;
|
||||
}
|
||||
|
||||
if (typeof payload === 'string') {
|
||||
return Buffer.from(payload, 'utf8'); // More explicit encoding
|
||||
}
|
||||
|
||||
// Convert to JSON string then encode
|
||||
return Buffer.from(JSON.stringify(payload), 'utf8');
|
||||
};
|
||||
Reference in New Issue
Block a user