-
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.
* Kafka workflow updates * Fixes to workflow
- Loading branch information
Showing
12 changed files
with
175 additions
and
108 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 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
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 |
|---|---|---|
| @@ -1,83 +1,111 @@ | ||
| import { Consumer, EachBatchPayload, Kafka, KafkaConfig, Message } from 'kafkajs'; | ||
| import logger from './winston'; | ||
| import { Consumer, EachBatchPayload, Kafka, KafkaConfig, Message } from 'kafkajs' | ||
| import logger from './winston' | ||
| import { WorkflowResult } from '../workflows/botswana/workflowHandler' | ||
|
|
||
| export type EachMessageCallback = (topic: string, partition: number, message: Message) => Promise<void>; | ||
| export type EachMessageCallback = ( | ||
| topic: string, | ||
| partition: number, | ||
| message: Message, | ||
| ) => Promise<WorkflowResult> | ||
|
|
||
| export class KafkaConsumerUtil { | ||
| private consumer: Consumer | null = null; | ||
| 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(); | ||
| this.consumer = await this.createConsumer() | ||
| } catch (err) { | ||
| console.error('Failed to initialize consumer:', err); | ||
| throw 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(); | ||
| const kafka = new Kafka(this.config) | ||
| const consumer = kafka.consumer({ | ||
| groupId: this.groupId, | ||
| sessionTimeout: 120000, // 2 minutes | ||
| heartbeatInterval: 30000, // 30 seconds | ||
| }) | ||
|
|
||
| await consumer.connect() | ||
|
|
||
| for (const topic of this.topics) { | ||
| await consumer.subscribe({ topic, fromBeginning: false }); | ||
| await consumer.subscribe({ topic, fromBeginning: false }) | ||
| } | ||
| return consumer; | ||
| return consumer | ||
| } | ||
|
|
||
| public async consumeTransactionally(eachMessageCallback: EachMessageCallback): Promise<void> { | ||
| if (!this.consumer) { | ||
| throw new Error('Consumer is not initialized.'); | ||
| 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; | ||
|
|
||
| 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(), | ||
| }); | ||
|
|
||
| const maxRetries = 6; | ||
| let retryCount = 0; | ||
| let retryDelay = 1000; | ||
|
|
||
| if (!isRunning() || isStale()) return | ||
|
|
||
| logger.info( | ||
| `Consumer | Recieved message from topic ${topic} on partition ${partition} with offset ${message.offset}`, | ||
| ) | ||
|
|
||
| const maxRetries = 2 | ||
| let retryCount = 0 | ||
| let retryDelay = 2000 | ||
| let res: WorkflowResult | null = null | ||
|
|
||
| while (retryCount < maxRetries) { | ||
| logger.info(`Processing message for ${topic} with retry count ${retryCount}...`) | ||
| try { | ||
| await eachMessageCallback(topic, partition, message); | ||
| resolveOffset(message.offset); | ||
| await heartbeat(); | ||
| break; // Break the loop if processing succeeds | ||
| } catch (error) { | ||
| logger.error(`Error processing message ${message.offset}: ${error}`); | ||
| retryCount++; | ||
| if (retryCount >= maxRetries) { | ||
| logger.error(`Max retries reached for message ${message.offset}, sending to dead letter queue or similar.`); | ||
| // TODO: handle with DLQ | ||
| break; | ||
| res = await eachMessageCallback(topic, partition, message) | ||
|
|
||
| if (res.success) { | ||
| logger.info(`Workflow result succeeded!`) | ||
| resolveOffset(message.offset) | ||
| await heartbeat() | ||
| break // Break the loop if processing succeeds } | ||
| } else { | ||
| logger.error(`Workflow result did not succeed: ${res.result}`) | ||
| } | ||
| await new Promise(resolve => setTimeout(resolve, retryDelay)); | ||
| retryDelay *= 2; // Double the delay for the next retry | ||
| await heartbeat(); // Important to call heartbeat to keep the session alive | ||
| } catch (error) { | ||
| logger.error(`Error processing message ${message.offset}: ${error}`) | ||
| } | ||
|
|
||
| // Otherwise, retry, both on error, and if the message is not processed successfully | ||
| retryCount++ | ||
| if (retryCount >= maxRetries) { | ||
| logger.error( | ||
| `Max retries reached for message ${message.offset}, sending to dead letter queue or similar.`, | ||
| ) | ||
| resolveOffset(message.offset) | ||
|
|
||
| // TODO: handle with DLQ | ||
| break | ||
| } | ||
| await new Promise(resolve => setTimeout(resolve, retryDelay)) | ||
| retryDelay *= 20 // Double the delay for the next retry | ||
| await heartbeat() // Important to call heartbeat to keep the session alive | ||
| } | ||
| } | ||
| }, | ||
| }); | ||
| }) | ||
| } | ||
|
|
||
| public async shutdown(): Promise<void> { | ||
| if (this.consumer) { | ||
| await this.consumer.disconnect(); | ||
| 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
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
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.