import { Injectable, OnApplicationShutdown, Inject, forwardRef, OnApplicationBootstrap } from '@nestjs/common'; import { Consumer, Kafka, KafkaConfig, SASLOptions } from 'kafkajs'; import axios from "axios"; import { v4 as uuidv4 } from 'uuid'; import { AsyncListnerDataService } from "src/AsyncListnerData/AsyncListnerData.service"; import { WinstonLoggerService } from 'src/loggers/winston-logger.service'; import { EngineService } from 'src/engine/engine.service'; import { RedisService } from "src/redis/redis.service"; const fs = require('fs') @Injectable() export class ConsumerAsyncService implements OnApplicationBootstrap, OnApplicationShutdown { private kafka?: Kafka; private readonly brokers: string[] = [process.env.KAFKA_HOST]; private isInitialized: boolean = false; private mainConsumer: Consumer | null = null; private isConsumerStarted = false; private readonly topicConfig = new Map(); private readonly topicState = new Map(); private readonly pausedPartitions = new Map(); private readonly partitionsConsumedConcurrently = Number(process.env.KAFKA_PARTITIONS_CONCURRENCY || 1); private readonly smoothThrottleEnabled = process.env.SMOOTH_THROTTLE_ENABLED === "true"; private readonly isRedisEnabled = process.env.ENABLE_REDIS === "true"; constructor( @Inject(forwardRef(() => EngineService)) private readonly engineService: EngineService, @Inject(forwardRef(() => AsyncListnerDataService)) private asyncListnerDataService: AsyncListnerDataService, private readonly redisService: RedisService, private readonly logger: WinstonLoggerService ) { if (process.env.ENABLE_KAFKA === 'true') { this.kafka = this.createKafkaClient(); this.logger.logInfo( `Kafka initialized with protocol: ${process.env.KAFKA_SECURITY_PROTOCOL || 'PLAINTEXT'}`); } } private createKafkaClient(): Kafka { const protocol = process.env.KAFKA_SECURITY_PROTOCOL?.toUpperCase() || 'PLAINTEXT'; const config: KafkaConfig = { brokers: this.brokers, connectionTimeout: 10000, requestTimeout: 60000, retry: { initialRetryTime: 300, retries: 8, }, }; switch (protocol) { case 'SSL': config.ssl = { rejectUnauthorized: true, ca: process.env.CA_CERT ? [fs.readFileSync(process.env.CA_CERT, 'utf8')] : undefined, cert: process.env.CERT_PEM ? fs.readFileSync(process.env.CERT_PEM, 'utf8') : undefined, key: process.env.KEY_PEM ? fs.readFileSync(process.env.KEY_PEM, 'utf8') : undefined, }; this.logger.logInfo('Kafka configured with SSL/mTLS',); break; case 'SASL_SSL': config.ssl = { rejectUnauthorized: true, ca: process.env.CA_CERT ? [fs.readFileSync(process.env.CA_CERT, 'utf8')] : undefined, }; config.sasl = { mechanism: (process.env.KAFKA_SASL_MECHANISM as | 'plain' | 'scram-sha-256' | 'scram-sha-512') || 'plain', username: process.env.KAFKA_USERNAME || '', password: process.env.KAFKA_PASSWORD || '', } as SASLOptions; this.logger.logInfo(`Kafka configured with SASL_SSL (${process.env.KAFKA_SASL_MECHANISM})`,); break; case 'PLAINTEXT': default: this.logger.logInfo('Kafka configured with PLAINTEXT',); break; } return new Kafka(config); } async onApplicationBootstrap(): Promise { try { await this.engineService.reloadListeners(); this.isInitialized = true; await this.reloadFirst(); } catch (e: any) { this.logger.logError( "In ConsumerAsyncService onApplicationBootstrap method Kafka connection issue", e.message ); } } private attachMainConsumerEvents(consumer: Consumer) { consumer.on(consumer.events.CONNECT, () => { if (this.mainConsumer !== consumer) return; this.logger.logInfo("[mainConsumer] CONNECT"); }); consumer.on(consumer.events.DISCONNECT, () => { if (this.mainConsumer !== consumer) return; this.logger.logInfo("[mainConsumer] DISCONNECT"); this.isConsumerStarted = false; this.mainConsumer = null; }); consumer.on(consumer.events.STOP, () => { if (this.mainConsumer !== consumer) return; this.logger.logInfo("[mainConsumer] STOP"); this.isConsumerStarted = false; }); consumer.on(consumer.events.CRASH, ({ payload }) => { if (this.mainConsumer !== consumer) return; this.logger.logError( "[mainConsumer] CRASH", payload?.error?.message || "unknown crash" ); this.isConsumerStarted = false; this.mainConsumer = null; }); } private async subscribeWithRetry( consumer: Consumer, topics: string[], options?: { maxRetries?: number; baseDelayMs?: number; } ): Promise<{ success: string[], failed: string[] }> { const maxRetries = options?.maxRetries ?? 2; const baseDelayMs = options?.baseDelayMs ?? 100; let failedTopics: string[] = []; const successfulTopics: string[] = []; // INITIAL PASS for (const topic of topics) { try { await consumer.subscribe({ topic, fromBeginning: false }); successfulTopics.push(topic); } catch (e: any) { failedTopics.push(topic); this.logger.logError(`[subscribe] ❌ ${topic}`, e.message); } } // RETRY LOOP for (let attempt = 1; attempt <= maxRetries && failedTopics.length > 0; attempt++) { const delay = baseDelayMs * attempt; this.logger.logInfo( `[subscribe] 🔁 retry=${attempt}, wait=${delay}ms, topics=${failedTopics.join(", ")}` ); await new Promise(res => setTimeout(res, delay)); const nextFailed: string[] = []; for (const topic of failedTopics) { try { await consumer.subscribe({ topic, fromBeginning: false }); successfulTopics.push(topic); } catch (e: any) { nextFailed.push(topic); } } failedTopics = nextFailed; } if (failedTopics.length > 0) { this.logger.logError( `[subscribeWithRetry] 🚫 Final failed topics: ${failedTopics.join(", ")}` ); } return { success: successfulTopics, failed: failedTopics }; } private getThrottleDecision(topicObj: any): { allow: boolean; pauseAfterCurrent: boolean; waitMs: number; } { const topic = topicObj.topic; const txType = topicObj.transaction_type; const txInterval = Number(topicObj.transaction_interval); const now = Date.now(); const intervalMs = this.getIntervalMs(txType); if (!this.topicState.has(topic)) { this.topicState.set(topic, { count: 0, windowStart: now, lastExecutionAt: 0, }); } const state = this.topicState.get(topic)!; if (now - state.windowStart >= intervalMs) { state.count = 0; state.windowStart = now; state.lastExecutionAt = 0; } if (txInterval <= 0) { return { allow: false, pauseAfterCurrent: false, waitMs: intervalMs, }; } const remainingWindowMs = Math.max(0, intervalMs - (now - state.windowStart)); if (this.smoothThrottleEnabled) { const spacingMs = Math.floor(intervalMs / txInterval); if ( state.lastExecutionAt && now - state.lastExecutionAt < spacingMs ) { const waitMs = spacingMs - (now - state.lastExecutionAt); return { allow: false, pauseAfterCurrent: false, waitMs, }; } } if (state.count >= txInterval) { return { allow: false, pauseAfterCurrent: false, waitMs: remainingWindowMs, }; } state.count += 1; state.lastExecutionAt = now; if (state.count >= txInterval) { return { allow: true, pauseAfterCurrent: true, waitMs: remainingWindowMs, }; } if (this.smoothThrottleEnabled) { const spacingMs = Math.floor(intervalMs / txInterval); return { allow: true, pauseAfterCurrent: true, waitMs: Math.min(spacingMs, remainingWindowMs), }; } return { allow: true, pauseAfterCurrent: false, waitMs: 0, }; } private async waitWithHeartbeat( waitMs: number, heartbeat: () => Promise ): Promise { const safeWaitMs = Math.max(0, waitMs); const endAt = Date.now() + safeWaitMs; while (Date.now() < endAt) { const remaining = endAt - Date.now(); const sleepMs = Math.min(1000, remaining); await new Promise(resolve => setTimeout(resolve, sleepMs)); await heartbeat(); } } private pauseTopic( topic: string, reason: 'THROTTLE' | 'ZERO_INTERVAL', resumeAfterMs?: number ): void { if (!this.mainConsumer) return; const existing = this.pausedPartitions.get(topic); if (existing?.resumeTimer) { clearTimeout(existing.resumeTimer); } this.logger.logInfo( `[pauseTopic] Pausing topic=${topic}, reason=${reason}, resumeAfter=${resumeAfterMs}` ); this.mainConsumer.pause([{ topic }]); const newState: { paused: boolean; reason: 'THROTTLE' | 'ZERO_INTERVAL' | null; resumeTimer?: NodeJS.Timeout; } = { paused: true, reason, }; if (resumeAfterMs !== undefined && resumeAfterMs > 0) { const timerId = setTimeout(() => { this.resumeTopic(topic); }, resumeAfterMs); newState.resumeTimer = timerId; this.logger.logInfo( `[pauseTopic] Timer SET for ${topic}, duration=${resumeAfterMs}` ); } this.pausedPartitions.set(topic, newState); } private resumeTopic(topic: string): void { if (!this.mainConsumer) return; const state = this.pausedPartitions.get(topic); if (!state?.paused) return; if (state.resumeTimer) { clearTimeout(state.resumeTimer); } this.logger.logInfo( `[resumeTopic] Resuming topic=${topic}, reason=${state.reason}` ); this.mainConsumer.resume([{ topic }]); this.pausedPartitions.set(topic, { paused: false, reason: null, }); this.logger.logInfo(`[ConsumerAsyncService][resumeTopic] ${topic} resumed`); } private sameTopicSet(a: string[], b: string[]): boolean { if (a.length !== b.length) return false; const aSorted = [...a].sort(); const bSorted = [...b].sort(); for (let i = 0; i < aSorted.length; i++) { if (aSorted[i] !== bSorted[i]) return false; } return true; } async reloadAsyncListeners(topicPauseList: string[]) { this.logger.logInfo( "[reloadAsyncListeners] START changed topics :: ", topicPauseList ); try { const previousTopics = [...this.topicConfig.keys()]; const asyncListnerData = await this.asyncListnerDataService.reloadAsyncListners(); const latestConfigMap = new Map(); for (const item of asyncListnerData) { latestConfigMap.set(item.topic, item); } const latestTopics = [...latestConfigMap.keys()]; const topicSetChanged = !this.sameTopicSet(previousTopics, latestTopics); const consumerMissing = !this.mainConsumer || !this.isConsumerStarted; this.logger.logInfo( `[reloadAsyncListeners] topicSetChanged=${topicSetChanged}, consumerMissing=${consumerMissing}` ); this.topicConfig.clear(); for (const [topic, config] of latestConfigMap.entries()) { this.topicConfig.set(topic, config); } for (const topic of latestTopics) { if (!this.topicState.has(topic)) { this.topicState.set(topic, { count: 0, windowStart: Date.now(), lastExecutionAt: 0, }); } } for (const topic of topicPauseList) { if (!this.topicConfig.has(topic)) continue; const newCfg = this.topicConfig.get(topic); const pausedState = this.pausedPartitions.get(topic); let state = this.topicState.get(topic); const newLimit = Number(newCfg?.transaction_interval); const txType = newCfg?.transaction_type; const intervalMs = this.getIntervalMs(txType); if (!state) { state = { count: 0, windowStart: Date.now(), lastExecutionAt: 0, }; this.topicState.set(topic, state); } if (pausedState?.resumeTimer) { clearTimeout(pausedState.resumeTimer); } if (isNaN(newLimit) || newLimit < 0) { this.logger.logInfo( `[reloadAsync] skipping ${topic} due to invalid config` ); continue; } if ( (txType === "Per Second" || txType === "Per Minute" || txType === "Per Hour") && newLimit === 0 ) { this.pauseTopic(topic, "ZERO_INTERVAL"); this.logger.logInfo( `[reloadAsync] Applied ZERO_INTERVAL hard stop immediately for ${topic}` ); continue; } const elapsed = Date.now() - state.windowStart; const remaining = Math.max(0, intervalMs - elapsed); if (state.count >= newLimit) { if (remaining > 0) { this.pauseTopic(topic, "THROTTLE", remaining); this.logger.logInfo( `[reloadAsync] ${topic} remains throttled. count=${state.count}, newLimit=${newLimit}, remaining=${remaining}` ); } else { this.resumeTopic(topic); state.count = 0; state.windowStart = Date.now(); state.lastExecutionAt = 0; this.logger.logInfo( `[reloadAsync] ${topic} window expired during reload. Reset and resumed` ); } } else { if (pausedState?.paused) { this.logger.logInfo( `[reload FIX] Resuming topic=${topic} because count=${state.count} < newLimit=${newLimit}` ); this.resumeTopic(topic); } this.logger.logInfo( `[reloadAsync] ${topic} can continue in current window. count=${state.count}, newLimit=${newLimit}, remainingAllowed=${newLimit - state.count}` ); } } if (!topicSetChanged && !consumerMissing) { this.logger.logInfo( "[reloadAsyncListeners] ONLY config changed, consumer healthy, no reconnect" ); return; } this.logger.logInfo( `[reloadAsyncListeners] Reconnecting main consumer. topicSetChanged=${topicSetChanged}, consumerMissing=${consumerMissing}` ); if (this.mainConsumer) { try { await this.mainConsumer.disconnect(); this.logger.logInfo("[reloadAsync] Main consumer disconnected"); } catch (e: any) { this.logger.logError("[reloadAsync] Error disconnecting main consumer", e.message); } finally { this.mainConsumer = null; this.isConsumerStarted = false; } } for (const [topic, state] of this.pausedPartitions.entries()) { if (state.resumeTimer) { clearTimeout(state.resumeTimer); } this.pausedPartitions.set(topic, { paused: false, reason: null, }); } this.mainConsumer = this.kafka.consumer({ groupId: process.env.KAFKA_GROUPID, sessionTimeout: 60000, heartbeatInterval: 6000, ...({ groupInstanceId: `${process.env.HOSTNAME || uuidv4()}` } as any) }); this.attachMainConsumerEvents(this.mainConsumer); await this.mainConsumer.connect(); this.logger.logInfo("[reloadAsync] Main consumer connected after topic change"); const { success: subscribedTopics } = await this.subscribeWithRetry(this.mainConsumer, latestTopics); if (subscribedTopics.length === 0) { this.logger.logError("[reloadAsync] No topics subscribed after reconnect"); return; } this.logger.logInfo(`[reloadAsync] Subscribed topics count=${subscribedTopics.length}`); for (const topic of subscribedTopics) { if (!this.topicState.has(topic)) { this.topicState.set(topic, { count: 0, windowStart: Date.now(), lastExecutionAt: 0, }); } } if (!this.isConsumerStarted) { await this.run(this.mainConsumer); this.isConsumerStarted = true; this.logger.logInfo("[reloadAsync] Main consumer started"); } for (const [topic, cfg] of this.topicConfig.entries()) { const txType2 = cfg?.transaction_type; const txInterval2 = Number(cfg?.transaction_interval); const state = this.topicState.get(topic); if (!state) continue; if ( (txType2 === "Per Second" || txType2 === "Per Minute" || txType2 === "Per Hour") && txInterval2 === 0 ) { this.pauseTopic(topic, "ZERO_INTERVAL"); this.logger.logInfo( `[reload reconnect FIX] Re-applied ZERO_INTERVAL pause for ${topic}` ); continue; } const intervalMs2 = this.getIntervalMs(txType2); const elapsed2 = Date.now() - state.windowStart; const remaining2 = Math.max(0, intervalMs2 - elapsed2); if (state.count >= txInterval2 && remaining2 > 0) { this.pauseTopic(topic, "THROTTLE", remaining2); } } this.logger.logInfo("[reloadAsyncListeners] SUCCESS"); } catch (e: any) { this.logger.logError( "[reloadAsyncListeners] ERROR:", e.message ); } } async pauseBeforeReload(topicList: string[]) { try { this.logger.logInfo( "[ConsumerAsyncService][pauseBeforeReload] Changed topics before reload :: ", topicList ); if (process.env.ENABLE_KAFKA !== "true") return; if (!this.mainConsumer || !this.isConsumerStarted) { this.logger.logInfo( "[ConsumerAsyncService][pauseBeforeReload] Main consumer not started, nothing to pause" ); return; } if (!topicList || topicList.length === 0) { this.logger.logInfo( "[ConsumerAsyncService][pauseBeforeReload] No changed topics received" ); return; } for (const topic of topicList) { const existing = this.pausedPartitions.get(topic); if (existing?.resumeTimer) { clearTimeout(existing.resumeTimer); } if (existing?.paused && existing.reason === "RELOAD") { continue; } try { this.mainConsumer.pause([{ topic }]); this.pausedPartitions.set(topic, { paused: true, reason: 'RELOAD', }); } catch (error: any) { this.logger.logError( `[ConsumerAsyncService][pauseBeforeReload] Failed to pause topic ${topic}`, error.message ); } } } catch (e: any) { this.logger.logError( "[ConsumerAsyncService][pauseBeforeReload] Kafka connection issue", e.message ); } } async reloadFirst() { try { if (process.env.ENABLE_KAFKA !== "true") return; const asyncListnerData = await this.asyncListnerDataService.reloadAsyncListners(); this.logger.logInfo(">>> reloadFirst START"); this.topicConfig.clear(); this.topicState.clear(); this.pausedPartitions.clear(); const topicList: string[] = []; for (const item of asyncListnerData) { this.topicConfig.set(item.topic, item); topicList.push(item.topic); this.topicState.set(item.topic, { count: 0, windowStart: Date.now(), lastExecutionAt: 0, }); } if (!this.mainConsumer) { this.mainConsumer = this.kafka.consumer({ groupId: process.env.KAFKA_GROUPID, sessionTimeout: 60000, heartbeatInterval: 6000, ...({ groupInstanceId: `${process.env.HOSTNAME || uuidv4()}` } as any) }); this.logger.logInfo(">>> connecting consumer"); this.attachMainConsumerEvents(this.mainConsumer); await this.mainConsumer.connect(); this.logger.logInfo("[ConsumerAsyncService][reloadFirst] Main consumer connected"); } this.logger.logInfo(">>> subscribing to all topics using retry"); const { success: subscribedTopics } = await this.subscribeWithRetry(this.mainConsumer, topicList); if (subscribedTopics.length === 0) { this.logger.logError("[reloadFirst] No topics subscribed. Not starting consumer."); return; } this.logger.logInfo( `[reloadFirst] Subscribed topics count=${subscribedTopics.length}` ); if (!this.isConsumerStarted) { this.logger.logInfo(">>> starting RUN()"); await this.run(this.mainConsumer); this.isConsumerStarted = true; this.logger.logInfo("[ConsumerAsyncService][reloadFirst] Main consumer started"); } this.logger.logInfo("[ConsumerAsyncService][reloadFirst] Reloaded async listeners successfully"); } catch (e: any) { this.logger.logError( "[ConsumerAsyncService][reloadFirst] Kafka Connection Issue:", e.message ); } } async run(consumer: Consumer) { try { await consumer.run({ autoCommit: false, partitionsConsumedConcurrently: this.partitionsConsumedConcurrently, eachMessage: async ({ topic, partition, message, heartbeat }) => { if (!this.isInitialized) { return; } const currentOffset = message.offset; const committedOffset = (Number(currentOffset) + 1).toString(); const commitOffset = async (reason: string) => { this.logger.logInfo( `[KAFKA_COMMIT] topic=${topic}, partition=${partition}, processedOffset=${currentOffset}, committedOffset=${committedOffset}, reason=${reason}` ); await consumer.commitOffsets([ { topic, partition, offset: committedOffset, }, ]); await heartbeat(); }; if (!message?.value) { this.logger.logInfo( `[KAFKA_EMPTY_MESSAGE] topic=${topic}, partition=${partition}, offset=${currentOffset}` ); await commitOffset("EMPTY_MESSAGE"); return; } const topicObj = this.topicConfig.get(topic); if (!topicObj) { this.logger.logInfo( `[KAFKA_NO_CONFIG] topic=${topic}, partition=${partition}, offset=${currentOffset}` ); await commitOffset("NO_TOPIC_CONFIG"); return; } const transactionType = topicObj.transaction_type; const transactionInterval = Number(topicObj.transaction_interval); const key = `${topic}-${partition}-${currentOffset}`; if(this.isRedisEnabled){ const currDate = new Date().toISOString(); await this.redisService.setKafkaKey(key,`PROCESSING-${currDate}`); } try { await heartbeat(); if (transactionType === "NA") { console.log("[KAFKA_PROCESSING][run][NA] ", key,new Date().toISOString() ); this.logger.logInfo( `[KAFKA_PROCESSING][NA] topic=${topic}, partition=${partition}, offset=${currentOffset}, txType=NA` ); const kafkakey = this.isRedisEnabled ? await this.redisService.getKafkaKey(key) : "PROCESSING"; if(kafkakey === "PROCESSING"){ const res = await this.hitEndPoint( topicObj.method, topicObj.triggerURL, message.value.toString(), topicObj.headers, key ); this.logger.logInfo("PROCESSED_NA messaged hit:::::", { topic, partition, offset: (Number(message.offset) + 1).toString(), message: res?.message }) } if(kafkakey && kafkakey !== "Error"){ await commitOffset("PROCESSED_NA"); } return; } if ( (transactionType === "Per Second" || transactionType === "Per Minute" || transactionType === "Per Hour") && transactionInterval === 0 ) { this.logger.logInfo( `[KAFKA_HARD_STOP] topic=${topic}, partition=${partition}, offset=${currentOffset}, txType=${transactionType}, txInterval=${transactionInterval}, action=PAUSE_NO_COMMIT` ); this.pauseTopic(topic, "ZERO_INTERVAL"); await heartbeat(); return; } let decision = this.getThrottleDecision(topicObj); while (!decision.allow) { this.logger.logInfo( `[KAFKA_THROTTLED_WAIT] topic=${topic}, partition=${partition}, offset=${currentOffset}, waitMs=${decision.waitMs}` ); if (decision.waitMs > 0) { this.pauseTopic(topic, "THROTTLE", decision.waitMs); await this.waitWithHeartbeat(decision.waitMs, heartbeat); } else { await heartbeat(); } /** * IMPORTANT: * Re-check throttle for the SAME Kafka message. * Do not return from this offset before processing. */ decision = this.getThrottleDecision(topicObj); } console.log("[KAFKA_PROCESSING][run] ", key , new Date().toISOString()); this.logger.logInfo( `[KAFKA_PROCESSING] topic=${topic}, partition=${partition}, offset=${currentOffset}, txType=${transactionType}, txInterval=${transactionInterval}` ); const kafkaKey = this.isRedisEnabled ? await this.redisService.getKafkaKey(key) : "PROCESSING"; this.logger.logInfo("kafka key before hitendpoint :: ",kafkaKey); if( kafkaKey === "PROCESSING"){ const res = await this.hitEndPoint( topicObj.method, topicObj.triggerURL, message.value.toString(), topicObj.headers, key ); this.logger.logInfo("messaged hit: ::::", { topic, partition, offset: (Number(message.offset) + 1).toString(), message: res.message }) } if(kafkaKey && kafkaKey !== "Error"){ await commitOffset("PROCESSED_THROTTLED_TOPIC"); } if (decision.pauseAfterCurrent && decision.waitMs > 0) { this.logger.logInfo( `[KAFKA_LIMIT_REACHED] topic=${topic}, partition=${partition}, offset=${currentOffset}, pauseMs=${decision.waitMs}` ); this.pauseTopic(topic, "THROTTLE", decision.waitMs); } await heartbeat(); } catch (err: any) { const msg = err?.message || "unknown error"; if ( msg.includes("rebalancing") || msg.includes("rejoin is needed") || msg.includes("coordinator is not aware") ) { this.logger.logInfo( `[KAFKA_REBALANCE] topic=${topic}, partition=${partition}, offset=${currentOffset}, action=NO_COMMIT_RETRY_LATER, message=${msg}` ); } else { this.logger.logError( `[ConsumerAsyncService][run][eachMessage] Error topic=${topic}, partition=${partition}, offset=${currentOffset}`, msg ); } } }, }); } catch (err: any) { this.logger.logError( "[ConsumerAsyncService][run] Kafka Connection Issue:", err.message ); } } getIntervalMs(transactionType: string): number { const match = transactionType.match(/^(\d+)\s*seconds?$/i); if (match) return parseInt(match[1], 10) * 1000; switch (transactionType) { case "Per Second": return 1000; case "Per Minute": return 60000; case "Per Hour": return 3600000; default: return 1; } } async onApplicationShutdown() { try { if (process.env.ENABLE_KAFKA !== "true") return; for (const [topic, state] of this.pausedPartitions.entries()) { if (state.resumeTimer) { clearTimeout(state.resumeTimer); } this.pausedPartitions.set(topic, { paused: false, reason: null, }); } if (this.mainConsumer) { await this.mainConsumer.disconnect(); this.mainConsumer = null; this.isConsumerStarted = false; this.logger.logInfo("[ConsumerAsyncService][onApplicationShutdown] Main consumer disconnected"); } } catch (error: any) { this.logger.logError( "[ConsumerAsyncService][onApplicationShutdown] Kafka shutdown issue:", error.message ); } } async hitEndPoint(method: string, url: string, data, headers,key) { try { console.log("Hitting endpoint : ", url); const transactionId = uuidv4(); if (method === "POST") { const converted = JSON.parse(data); const updatedHeaders = { ...headers, headers: { ...(headers?.headers || {}), "circuit-transactionid": transactionId, "x-circuit-async": key }, }; converted.circuit_Transaction_id = transactionId; try { const res = await axios.post(url, converted, updatedHeaders); return res.data; } catch (error : any) { this.logger.logError("Error in hitEnpoint POST::::", error.message); } } else if (method === "GET") { const updatedHeaders = { ...headers, headers: { ...(headers?.headers || {}), "circuit-transactionid": transactionId, "x-circuit-async": key }, }; try { const res = await axios.get(url, updatedHeaders); return res.data; } catch (error: any) { this.logger.logError("Error in hitEnpoint GET::::", error.message) } } else if (method === "PUT") { const updatedHeaders = { ...headers, headers: { ...(headers?.headers || {}), "x-circuit-async": true }, }; const converted = JSON.parse(data); converted.circuit_Transaction_id = transactionId; try { const res = await axios.put(url, converted, updatedHeaders); return res.data; } catch (error: any) { this.logger.logError("Error in hitEnpoint PUT::::", error.message) } } else if (method === "DELETE") { const updatedHeaders = { ...headers, headers: { ...(headers?.headers || {}), "circuit-transactionid": transactionId, "x-circuit-async": true }, }; try { const res = await axios.delete(url, updatedHeaders).catch(() => { }); return res; } catch (error: any) { this.logger.logError("Error in hitEnpoint PUT::::", error.message) } } } catch (err: any) { this.logger.logError( "[ConsumerAsyncService][hitEndPoint] !! Error message !! : ", err.message ); } } }

myconsumer