Kafka

The Kafka plugin creates named KafkaJS clients and manages a producer and consumer for each connection. It passes each connection's Kafka configuration to KafkaJS and connects configured producers and consumers during plugin registration. See the KafkaJS documentation for Kafka concepts and the full client API.

Install

Install the Kafka plugin in your project:

npm install @axisparkjs/kafka

Configure connections

Register KafkaPlugin with options created by KafkaPluginOptionsFactory. Each connection needs a unique name and a KafkaJS config, including at least one broker:

app.ts

import { AxiSparkFactory } from '@axisparkjs/core';
import { KafkaPlugin, KafkaPluginOptionsFactory } from '@axisparkjs/kafka';

export const app = AxiSparkFactory.create({
    name: 'My application',
    basePath: __dirname
});

app.use(
    KafkaPlugin,
    KafkaPluginOptionsFactory.create({
        connections: [
            {
                name: 'ORDERS',
                config: {
                    clientId: 'orders-service',
                    brokers: ['localhost:9092']
                },
                producerConfig: {
                    idempotent: true
                },
                consumerConfig: {
                    groupId: 'orders-workers'
                }
            },
            {
                name: 'EVENTS',
                config: {
                    clientId: 'events-service',
                    brokers: ['broker-a:9092', 'broker-b:9092'],
                    ssl: true
                },
                autoInitialize: false
            }
        ]
    })
);

config is passed to the KafkaJS Kafka constructor. producerConfig and consumerConfig are passed to kafka.producer() and kafka.consumer(), respectively. The plugin supplies a default consumer group ID of <connection-name-lowercase>-group; setting consumerConfig.groupId overrides it. See KafkaJS client configuration, producer, and consumer docs for available options and APIs.

By default, the plugin creates and connects both a producer and consumer for every connection during registration. Set autoInitialize: false to create them lazily when first injected. To initialize only one client automatically, configure the flags separately:

{
    name: 'PUBLISH_ONLY',
    config: {
        clientId: 'publisher',
        brokers: ['localhost:9092']
    },
    autoInitialize: {
        autoInitializeProducers: true,
        autoInitializeConsumers: false
    }
}

Connection names must be unique. The manager connects initialized clients during plugin registration and disconnects them when the plugin stops. Initialization errors are logged; the failed producer or consumer is not registered.

Inject a producer or consumer

Use @InjectKafkaProducer(name) or @InjectKafkaConsumer(name) to inject the corresponding KafkaJS client for a named connection:

import { Injectable } from '@axisparkjs/di';
import { InjectKafkaProducer } from '@axisparkjs/kafka';
import type { Producer } from 'kafkajs';

@Injectable()
export class OrderPublisher {
    constructor(
        @InjectKafkaProducer('ORDERS')
        private readonly producer: Producer
    ) {}

    async publish(order: { id: string; status: string }): Promise<void> {
        await this.producer.send({
            topic: 'orders',
            messages: [{ key: order.id, value: JSON.stringify(order) }]
        });
    }
}

A consumer is connected by the plugin but still needs a subscription and a run handler:

import { Injectable } from '@axisparkjs/di';
import { InjectKafkaConsumer } from '@axisparkjs/kafka';
import type { Consumer } from 'kafkajs';

@Injectable()
export class OrderListener {
    constructor(
        @InjectKafkaConsumer('ORDERS')
        private readonly consumer: Consumer
    ) {}

    async start(): Promise<void> {
        await this.consumer.subscribe({ topic: 'orders' });
        await this.consumer.run({
            eachMessage: async ({ message }) => {
                console.log(message.value?.toString());
            }
        });
    }
}

With autoInitialize: false, the plugin creates and connects the requested client when its injection token is first resolved. The connection manager tracks it and disconnects it on shutdown.

Inject a Kafka client

Use @InjectKafkaConnection(name) when you need the underlying KafkaJS Kafka client—for example, to create an additional producer that you manage yourself:

import { InjectKafkaConnection } from '@axisparkjs/kafka';
import type { Kafka } from 'kafkajs';

export class AdminService {
    constructor(
        @InjectKafkaConnection('ORDERS')
        private readonly kafka: Kafka
    ) {}

    createProducer() {
        return this.kafka.producer();
    }
}

The plugin only manages the producer and consumer it creates for each connection. If you create additional KafkaJS clients yourself, connect and disconnect them in your service. The corresponding injection tokens are KAFKA_CONNECTION_<name>, KAFKA_PRODUCER_<name>, and KAFKA_CONSUMER_<name>. The decorators are the recommended way to resolve them.

Use the connection manager

Inject KafkaConnectionManager to look up a named client, producer, or consumer, or to retrieve copies of the managed maps:

import { KafkaConnectionManager } from '@axisparkjs/kafka';

export class ConnectionInspector {
    constructor(private readonly connections: KafkaConnectionManager) {}

    inspect(): void {
        const kafka = this.connections.getConnection('ORDERS');
        const producer = this.connections.getProducer('ORDERS');
        const consumer = this.connections.getConsumer('ORDERS');

        console.log(kafka, producer, consumer);
        console.log([...this.connections.getAllConnections().keys()]);
    }
}

The getConnection, getProducer, and getConsumer methods return undefined when the name is unknown or the client has not been initialized. Each getAll... method returns a new map, so changing that map does not change the manager's internal state.