Class KafkaUtils

Utility class for managing Kafka consumers and producers. This class should not be instantiated.

Index

Constructors

Methods

disconnectFromKafka initializeConsumer initializeProducer

Constructors

constructor

new KafkaUtils(): KafkaUtils
Returns KafkaUtils
- Defined in kafka/kafkaUtils.ts:11

Methods

`Static`disconnectFromKafka

disconnectFromKafka(
    kafka: Kafka,
    consumers: Consumer[],
    producers: Producer[],
    app: FastifyInstance<
        RawServerDefault,
        IncomingMessage,
        ServerResponse<IncomingMessage>,
        FastifyBaseLogger,
        FastifyTypeProviderDefault,
    >,
): Promise<void>
Disconnects the consumer and producer from Kafka.
Parameters
- kafka: Kafka
  The Kafka instance to disconnect from.
- consumers: Consumer[]
- producers: Producer[]
- app: FastifyInstance<
      RawServerDefault,
      IncomingMessage,
      ServerResponse<IncomingMessage>,
      FastifyBaseLogger,
      FastifyTypeProviderDefault,
  >
  The Fastify instance used for logging.
Returns Promise<void>
A promise that resolves when both the consumer and producer are disconnected.
Throws
Will log an error if there is an issue disconnecting the consumer or producer.
Example
```
await KafkaUtils.disconnectFromKafka(kafka, consumers, producers, app);
```
- Defined in kafka/kafkaUtils.ts:81

`Static`initializeConsumer

initializeConsumer(
    kafka: Kafka,
    kafkaConfig: KafkaConfig,
    topic: string,
): Promise<Consumer>
Initializes a Kafka consumer with the specified configuration and subscribes to the given topic.
Parameters
- kafka: Kafka
  An instance of the Kafka client.
- kafkaConfig: KafkaConfig
  Configuration settings for the Kafka consumer, including the groupId.
- topic: string
  The topic to which the consumer should subscribe.
Returns Promise<Consumer>
A promise that resolves to the initialized Kafka consumer.
Example
```
const consumer = await KafkaUtils.initializeConsumer(kafka, kafkaConfig, 'my-topic');
consumer.run({
 eachMessage: async ({ topic, partition, message }) => {
  console.log({
   value: message.value.toString(),
 });
}
});
```
- Defined in kafka/kafkaUtils.ts:35

`Static`initializeProducer

initializeProducer(kafka: Kafka): Promise<Producer>
Initializes and connects a Kafka producer.
Parameters
- kafka: Kafka
  The Kafka instance to use for creating the producer.
Returns Promise<Producer>
A promise that resolves to the connected Kafka producer.
Example
```
const producer = await KafkaUtils.initializeProducer(kafka);
await producer.send({
topic: 'my-topic',
messages: [
 { value: 'Hello KafkaJS user!' },
],
});
```
- Defined in kafka/kafkaUtils.ts:59