-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Update kafka consumer and producer logic (#78)
* Kafka update * Kafka fixes * Consumer fixes * Fixed * Cleanup
- Loading branch information
Showing
8 changed files
with
304 additions
and
83 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| import { Kafka, Consumer, EachBatchPayload, Transaction, Message, KafkaConfig, EachMessagePayload } from 'kafkajs'; | ||
| import logger from './winston'; | ||
|
|
||
| export type EachMessageCallback = (topic: string, partition: number, message: Message) => Promise<void>; | ||
|
|
||
| export class KafkaConsumerUtil { | ||
| private consumer: Consumer | null = null; | ||
|
|
||
| constructor(private config: KafkaConfig, private topics: string[], private groupId: string) {} | ||
|
|
||
| public async init(): Promise<void> { | ||
| try { | ||
| this.consumer = await this.createConsumer(); | ||
| } catch (err) { | ||
| console.error('Failed to initialize consumer:', err); | ||
| throw err; | ||
| } | ||
| } | ||
|
|
||
| private async createConsumer(): Promise<Consumer> { | ||
| const kafka = new Kafka(this.config); | ||
| const consumer = kafka.consumer({ groupId: this.groupId }); | ||
| await consumer.connect(); | ||
| for (const topic of this.topics) { | ||
| await consumer.subscribe({ topic, fromBeginning: false }); | ||
| } | ||
| return consumer; | ||
| } | ||
|
|
||
| public async consumeTransactionally(eachMessageCallback: EachMessageCallback): Promise<void> { | ||
| if (!this.consumer) { | ||
| throw new Error('Consumer is not initialized.'); | ||
| } | ||
|
|
||
| await this.consumer.run({ | ||
| eachBatchAutoResolve: false, | ||
| eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning, isStale }: EachBatchPayload) => { | ||
| const { topic, partition } = batch; | ||
|
|
||
| for (const message of batch.messages) { | ||
| if (!isRunning() || isStale()) return; | ||
|
|
||
| logger.info({ | ||
| topic, | ||
| partition, | ||
| offset: message.offset, | ||
| value: message.value?.toString(), | ||
| }); | ||
|
|
||
| await eachMessageCallback(topic, partition, message) | ||
|
|
||
| resolveOffset(message.offset); | ||
| await heartbeat(); | ||
| } | ||
| }, | ||
| }); | ||
| } | ||
|
|
||
| public async shutdown(): Promise<void> { | ||
| if (this.consumer) { | ||
| await this.consumer.disconnect(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| import { Kafka, KafkaConfig, Producer, ProducerConfig, ProducerRecord, Transaction } from 'kafkajs'; | ||
| import config from './config' | ||
| import logger from './winston'; | ||
|
|
||
| type DeliveryReportCallback = (report: any) => void; | ||
|
|
||
| /** | ||
| * KafkaUtil class provides utility functions to interact with Kafka producer. | ||
| */ | ||
| export class KafkaProducerUtil { | ||
| private producer: Producer | null = null; | ||
|
|
||
| /** | ||
| * Creates an instance of KafkaUtil. | ||
| * @param {KafkaConfig} config - Configuration object for Kafka producer. | ||
| * @param {DeliveryReportCallback} onDeliveryReport - Callback function to handle delivery reports. | ||
| */ | ||
| constructor(private config: KafkaConfig, private onDeliveryReport: DeliveryReportCallback) {} | ||
|
|
||
| /** | ||
| * Initializes Kafka producer. | ||
| * @returns {Promise<void>} Promise that resolves when producer is initialized. | ||
| * @throws {Error} If producer initialization fails. | ||
| */ | ||
| public async init(): Promise<void> { | ||
| try { | ||
| this.producer = await this.createProducer(); | ||
| } catch (err) { | ||
| console.error('Failed to initialize producer:', err); | ||
| throw err; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Creates Kafka producer. | ||
| * @returns {Promise<Producer>} Promise that resolves with Kafka producer instance. | ||
| */ | ||
| private async createProducer(): Promise<Producer> { | ||
| logger.info('Creating Kafka producer...'); | ||
| const kafka = new Kafka(this.config); | ||
| const producer = kafka.producer({transactionalId: 'shr-producer-transaction', idempotent: true, maxInFlightRequests: 1}); | ||
| await producer.connect(); | ||
| return producer; | ||
| } | ||
|
|
||
| /** | ||
| * Sends message using transaction. | ||
| * @param {ProducerRecord[]} records - Array of producer records to send. | ||
| * @returns {Promise<void>} Promise that resolves when message is sent transactionally. | ||
| * @throws {Error} If producer is not initialized or transaction fails. | ||
| */ | ||
| public async sendMessageTransactionally(records: ProducerRecord[]): Promise<void> { | ||
| if (!this.producer) { | ||
| logger.error('Producer is not initialized.') | ||
| throw new Error('Producer is not initialized.'); | ||
| } | ||
|
|
||
| const transaction: Transaction = await this.producer.transaction(); | ||
| try { | ||
| logger.info('Sending the following records transactionally:'); | ||
| logger.info(JSON.stringify(records, null, 2)); | ||
| for (const record of records) { | ||
| await transaction.send(record); | ||
| } | ||
| await transaction.commit(); | ||
| this.onDeliveryReport({ status: 'committed' }); | ||
| } catch (err) { | ||
| await transaction.abort(); | ||
| this.onDeliveryReport({ status: 'aborted' }); | ||
| throw err; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Gracefully shuts down Kafka producer. | ||
| * @returns {Promise<void>} Promise that resolves when producer is disconnected. | ||
| */ | ||
| public async shutdown(): Promise<void> { | ||
| logger.info('Shutting down Kafka producer...'); | ||
| if (this.producer) { | ||
| await this.producer.disconnect(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.