Function disconnectKafka

packages/queue/src/kafka.ts:232–253

()

};

export const disconnectKafka = async (): Promise<void> => {
  const tasks: Promise<unknown>[] = [];
  for (const c of consumers) {
    tasks.push(
      c.disconnect().catch((err) => {
        kafkaLogger.error({ err }, 'kafka consumer disconnect error');
      })
    );
  }
  consumers.clear();
  if (producer) {
    const p = producer;
    producer = null;
    producerConnectPromise = null;
    tasks.push(
      p.disconnect().catch((err) => {
        kafkaLogger.error({ err }, 'kafka producer disconnect error');
      })
    );
  }
  await Promise.all(tasks);
};