Merge pull request #32078 from VELUX/feature/events-kafka
feat(events): add support for publishing to kafka and multiple instances
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1,5 @@
|
||||
---
|
||||
'@backstage/plugin-events-backend-module-kafka': minor
|
||||
---
|
||||
|
||||
Added support for multiple named instances in `kafkaConsumingEventPublisher` configuration. The previous single configuration format is still supported for backward compatibility.
|
||||
@@ -0,0 +1,7 @@
|
||||
---
|
||||
'@backstage/plugin-events-backend-module-kafka': minor
|
||||
---
|
||||
|
||||
Added `KafkaPublishingEventConsumer` to support sending Backstage events to Kafka topics.
|
||||
|
||||
This addition enables Backstage to publish events to external Kafka systems, complementing the existing ability to receive events from Kafka. This allows for better integration with external systems that rely on Kafka for event streaming.
|
||||
@@ -2,65 +2,107 @@
|
||||
|
||||
Welcome to the `events-backend-module-kafka` backend module!
|
||||
|
||||
This package is a module for the `events-backend` backend plugin and extends the events system with an `KafkaConsumingEventPublisher.`
|
||||
This package is a module for the `events-backend` backend plugin and extends the events system with a `KafkaConsumingEventPublisher` and `KafkaPublishingEventConsumer`
|
||||
|
||||
This event publisher will allow you to receive events from an Kafka queue and will publish these to the used `EventsService` implementation.
|
||||
This module provides two-way integration with Kafka:
|
||||
|
||||
- **KafkaConsumingEventPublisher**: Receives events from Kafka queues and publishes them to the Backstage events system
|
||||
- **KafkaPublishingEventConsumer**: Consumes events from Backstage and publishes them to Kafka queues
|
||||
|
||||
## Configuration
|
||||
|
||||
To set up Kafka queues, you need to configure the following values:
|
||||
To set up Kafka integration, you need to configure one or both of the following components:
|
||||
|
||||
### KafkaConsumingEventPublisher Configuration
|
||||
|
||||
To receive events from Kafka queues and publish them to Backstage:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaConsumingEventPublisher:
|
||||
clientId: your-client-id # (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
|
||||
brokers: # (Required) List of brokers in the Kafka cluster to connect to.
|
||||
- broker1
|
||||
- broker2
|
||||
topics:
|
||||
- topic: 'backstage.topic' # (Required) Replace with actual topic name as expected by subscribers
|
||||
kafka:
|
||||
topics: # (Required) The Kafka topics to subscribe to.
|
||||
- topic1
|
||||
groupId: your-group-id # (Required) The GroupId to be used by the topic consumers.
|
||||
production: # Instance name, will be included in logs
|
||||
clientId: your-client-id # (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
|
||||
brokers: # (Required) List of brokers in the Kafka cluster to connect to.
|
||||
- broker1
|
||||
- broker2
|
||||
topics:
|
||||
- topic: 'backstage.topic' # (Required) Replace with actual topic name as expected by subscribers
|
||||
kafka:
|
||||
topics: # (Required) The Kafka topics to subscribe to.
|
||||
- topic1
|
||||
groupId: your-group-id # (Required) The GroupId to be used by the topic consumers.
|
||||
```
|
||||
|
||||
### KafkaPublishingEventConsumer Configuration
|
||||
|
||||
To publish events from Backstage to Kafka queues, you can configure the `KafkaPublishingEventConsumer`:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaPublishingEventConsumer:
|
||||
production: # Instance name, will be included in logs
|
||||
clientId: your-client-id # (Required) Client ID used by Backstage to identify when connecting to the Kafka cluster.
|
||||
brokers: # (Required) List of brokers in the Kafka cluster to connect to.
|
||||
- broker1
|
||||
- broker2
|
||||
topics:
|
||||
- topic: 'catalog.entity.created' # (Required) The Backstage topic to consume from
|
||||
kafka:
|
||||
topic: kafka-topic-name # (Required) The Kafka topic to publish to
|
||||
```
|
||||
|
||||
For a complete list of all available fields that can be configured, refer to the [config.d.ts file](./config.d.ts).
|
||||
|
||||
### Optional SSL Configuration
|
||||
|
||||
If your Kafka cluster requires SSL, you can configure it as follows:
|
||||
If your Kafka cluster requires SSL, you can configure it for both `kafkaConsumingEventPublisher` and `kafkaPublishingEventConsumer` instances:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaConsumingEventPublisher:
|
||||
ssl:
|
||||
rejectUnauthorized: true # (Optional) If true, the server certificate is verified against the list of supplied CAs.
|
||||
ca: [path/to/ca-cert] # (Optional) Array of trusted certificates in PEM format.
|
||||
key: path/to/client-key # (Optional) Private key in PEM format.
|
||||
cert: path/to/client-cert # (Optional) Public x509 certificate in PEM format.
|
||||
production:
|
||||
# ... other configuration ...
|
||||
ssl:
|
||||
rejectUnauthorized: true # (Optional) If true, the server certificate is verified against the list of supplied CAs.
|
||||
ca: [path/to/ca-cert] # (Optional) Array of trusted certificates in PEM format.
|
||||
key: path/to/client-key # (Optional) Private key in PEM format.
|
||||
cert: path/to/client-cert # (Optional) Public x509 certificate in PEM format.
|
||||
kafkaPublishingEventConsumer:
|
||||
production:
|
||||
# ... other configuration ...
|
||||
ssl:
|
||||
# Same SSL configuration options as above
|
||||
```
|
||||
|
||||
### Optional SASL Authentication Configuration
|
||||
|
||||
If your Kafka cluster requires `SASL` authentication, you can configure it as follows:
|
||||
If your Kafka cluster requires SASL authentication, you can configure it for both components:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
kafka:
|
||||
kafkaConsumingEventPublisher:
|
||||
sasl:
|
||||
mechanism: 'plain' # SASL mechanism ('plain', 'scram-sha-256' or 'scram-sha-512')
|
||||
username: your-username # SASL username
|
||||
password: your-password # SASL password
|
||||
production:
|
||||
# ... other configuration ...
|
||||
sasl:
|
||||
mechanism: 'plain' # SASL mechanism ('plain', 'scram-sha-256' or 'scram-sha-512')
|
||||
username: your-username # SASL username
|
||||
password: your-password # SASL password
|
||||
kafkaPublishingEventConsumer:
|
||||
production:
|
||||
# ... other configuration ...
|
||||
sasl:
|
||||
# Same SASL configuration options as above
|
||||
```
|
||||
|
||||
This section includes optional `SSL` and `SASL` authentication configuration for enhanced security.
|
||||
These SSL and SASL configurations apply to both Kafka components and provide enhanced security for your Kafka connections.
|
||||
|
||||
## Installation
|
||||
|
||||
|
||||
+521
-153
@@ -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,548 @@ export interface Config {
|
||||
kafka?: {
|
||||
/**
|
||||
* Configuration for KafkaConsumingEventPublisher
|
||||
*
|
||||
* Supports either:
|
||||
* 1. Single configuration object (legacy format)
|
||||
* 2. Multiple named instances as a record where each key is a unique name for the Kafka instance
|
||||
*/
|
||||
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[];
|
||||
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 */
|
||||
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;
|
||||
};
|
||||
password: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Optional retry connection parameters.
|
||||
*/
|
||||
retry: {
|
||||
/**
|
||||
* (Optional) Maximum wait time for a retry
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
maxRetryTime: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Initial value used to calculate the retry (This is still randomized following the randomization factor)
|
||||
* Default: 300 ms.
|
||||
*/
|
||||
initialRetryTime: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Randomization factor
|
||||
* Default: 0.2.
|
||||
*/
|
||||
factor: number;
|
||||
|
||||
/**
|
||||
* (Optional) Exponential factor
|
||||
* Default: 2.
|
||||
*/
|
||||
multiplier: number;
|
||||
|
||||
/**
|
||||
* (Optional) Max number of retries per call
|
||||
* Default: 5.
|
||||
*/
|
||||
retries: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* (Optional) Timeout for authentication requests.
|
||||
* Default: 10000 ms.
|
||||
*/
|
||||
authenticationTimeout: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful connection.
|
||||
* Default: 1000 ms.
|
||||
*/
|
||||
connectionTimeout: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) Time to wait for a successful request.
|
||||
* Default: 30000 ms.
|
||||
*/
|
||||
requestTimeout: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false.
|
||||
* Default: true
|
||||
*/
|
||||
enforceRequestTimeout: boolean;
|
||||
|
||||
/**
|
||||
* Contains a object per topic for which an Kafka queue
|
||||
* should be used as source of events.
|
||||
*/
|
||||
topics: Array<{
|
||||
/**
|
||||
* (Required) The Backstage topic to publish to
|
||||
*/
|
||||
topic: string;
|
||||
/**
|
||||
* (Required) KafkaConsumer-related configuration.
|
||||
*/
|
||||
kafka: {
|
||||
/**
|
||||
* (Required) The Kafka topics to subscribe to
|
||||
* Optional retry connection parameters.
|
||||
*/
|
||||
topics: string[];
|
||||
/**
|
||||
* (Required) The GroupId to be used by the topic consumers
|
||||
*/
|
||||
groupId: string;
|
||||
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 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) 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.
|
||||
*/
|
||||
sessionTimeout: HumanDuration | string;
|
||||
requestTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (Optional) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
|
||||
* Default: 60000 ms.
|
||||
* (Optional) The request timeout can be disabled by setting enforceRequestTimeout to false.
|
||||
* Default: true
|
||||
*/
|
||||
rebalanceTimeout: HumanDuration | string;
|
||||
enforceRequestTimeout?: boolean;
|
||||
|
||||
/**
|
||||
* (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.
|
||||
* Contains an object per topic for which a Kafka queue
|
||||
* should be used as source of events.
|
||||
*/
|
||||
heartbeatInterval: HumanDuration | string;
|
||||
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) 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) 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 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) The maximum time that the coordinator will wait for each member to rejoin when rebalancing the group
|
||||
* Default: 60000 ms.
|
||||
*/
|
||||
rebalanceTimeout?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (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) 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) Maximum amount of bytes to accumulate in the response. Supported by Kafka >= 0.10.1.0
|
||||
* Default: 10485760 (10MB)
|
||||
*/
|
||||
maxBytes: 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).
|
||||
*/
|
||||
metadataMaxAge?: HumanDuration | string;
|
||||
|
||||
/**
|
||||
* (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;
|
||||
/**
|
||||
* (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;
|
||||
};
|
||||
}>;
|
||||
}
|
||||
| {
|
||||
[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 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;
|
||||
};
|
||||
};
|
||||
}>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
|
||||
@@ -6,6 +6,6 @@
|
||||
import { BackendFeature } from '@backstage/backend-plugin-api';
|
||||
|
||||
// @public
|
||||
const eventsModuleKafkaConsumingEventPublisher: BackendFeature;
|
||||
export default eventsModuleKafkaConsumingEventPublisher;
|
||||
const _default: BackendFeature;
|
||||
export default _default;
|
||||
```
|
||||
|
||||
+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, env.logger);
|
||||
|
||||
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,354 @@
|
||||
/*
|
||||
* 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';
|
||||
import { mockServices } from '@backstage/backend-test-utils';
|
||||
|
||||
const mockLogger = mockServices.logger.mock();
|
||||
|
||||
describe('readConsumerConfig', () => {
|
||||
it('not configured', () => {
|
||||
const publisherConfigs = readConsumerConfig(
|
||||
new ConfigReader({}),
|
||||
mockLogger,
|
||||
);
|
||||
|
||||
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, mockLogger);
|
||||
|
||||
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, mockLogger);
|
||||
|
||||
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, mockLogger);
|
||||
|
||||
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);
|
||||
});
|
||||
|
||||
it('single instance configuration (legacy format)', () => {
|
||||
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 = readConsumerConfig(config, mockLogger);
|
||||
|
||||
expect(publisherConfigs).toBeDefined();
|
||||
expect(Array.isArray(publisherConfigs)).toBe(true);
|
||||
expect(publisherConfigs).toHaveLength(1);
|
||||
|
||||
const defaultConfig = publisherConfigs[0];
|
||||
expect(defaultConfig.instance).toBe('default');
|
||||
expect(defaultConfig.kafkaConsumerConfigs.length).toBe(2);
|
||||
|
||||
expect(defaultConfig.kafkaConfig.clientId).toEqual('backstage-events');
|
||||
expect(defaultConfig.kafkaConfig.brokers).toEqual([
|
||||
'kafka1:9092',
|
||||
'kafka2:9092',
|
||||
]);
|
||||
|
||||
expect(defaultConfig.kafkaConsumerConfigs).toEqual([
|
||||
{
|
||||
backstageTopic: 'fake1',
|
||||
consumerConfig: {
|
||||
groupId: 'my-group',
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: ['topic-A'],
|
||||
},
|
||||
},
|
||||
{
|
||||
backstageTopic: 'fake2',
|
||||
consumerConfig: {
|
||||
groupId: 'my-group',
|
||||
},
|
||||
consumerSubscribeTopics: {
|
||||
topics: ['topic-B'],
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
// Verify deprecation warning was logged
|
||||
expect(mockLogger.warn).toHaveBeenCalledWith(
|
||||
'Legacy single config format detected at events.modules.kafka.kafkaConsumingEventPublisher.',
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,116 @@
|
||||
/*
|
||||
* 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';
|
||||
import { LoggerService } from '@backstage/backend-plugin-api';
|
||||
|
||||
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';
|
||||
|
||||
const processSinglePublisher = (
|
||||
instanceName: string,
|
||||
publisherConfig: Config,
|
||||
): KafkaConsumingEventPublisherConfig => {
|
||||
return {
|
||||
instance: instanceName,
|
||||
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'),
|
||||
},
|
||||
};
|
||||
}),
|
||||
};
|
||||
};
|
||||
|
||||
export const readConsumerConfig = (
|
||||
config: Config,
|
||||
logger: LoggerService,
|
||||
): KafkaConsumingEventPublisherConfig[] => {
|
||||
const publishersConfig = config.getOptionalConfig(CONFIG_PREFIX_PUBLISHER);
|
||||
|
||||
// Check for legacy single publisher format
|
||||
if (publishersConfig?.getOptionalString('clientId')) {
|
||||
logger.warn(
|
||||
'Legacy single config format detected at events.modules.kafka.kafkaConsumingEventPublisher.',
|
||||
);
|
||||
return [
|
||||
processSinglePublisher(
|
||||
'default', // use `default` as instance name for legacy single config
|
||||
publishersConfig,
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
return (
|
||||
publishersConfig
|
||||
?.keys()
|
||||
?.map(publisherKey =>
|
||||
processSinglePublisher(
|
||||
publisherKey,
|
||||
publishersConfig.getConfig(publisherKey),
|
||||
),
|
||||
) ?? []
|
||||
);
|
||||
};
|
||||
@@ -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