← Back to the section

Kafka messages are lost in two cases: if the consumer is killed before it managed to commit the processed message, or if the producer is killed before it managed to send the accumulated data to the broker. NestJS does not protect against this automatically — you need to manage the shutdown explicitly.

Why simply turning off is not enough

When the application receives a termination signal (SIGTERM), it starts stopping. If the consumer is processing another batch of messages at that moment, there are two scenarios:

  • the consumer is killed immediately — the offset is not committed, and on the next start the same messages arrive again (a repeat);
  • the consumer waits for the current batch to finish processing, then commits the offset, and only then closes — the data is preserved.

The second scenario is called a graceful shutdown. For it to happen, you need to explicitly call consumer.disconnect() in the beforeApplicationShutdown hook.

How to stop the consumer

In kafkajs the consumer.disconnect() method does three things: sends the broker a LeaveGroup command, waits for the current eachMessage or eachBatch to finish, and closes the connections. This is exactly why the offset has time to commit before closing.

@Injectable()
export class OrderConsumer implements BeforeApplicationShutdown {
  private readonly consumer: Consumer;

  constructor(private readonly kafka: Kafka) {
    this.consumer = kafka.consumer({ groupId: 'order-service-confirmations' });
  }

  async onModuleInit(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'orders.confirmed', fromBeginning: false });
    await this.consumer.run({
      eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary, isRunning }) => {
        for (const message of batch.messages) {
          if (!isRunning()) break;
          await this.handle(message);
          resolveOffset(message.offset);
          await commitOffsetsIfNecessary();
        }
      },
    });
  }

  async beforeApplicationShutdown(): Promise<void> {
    await Promise.race([
      this.consumer.disconnect(),
      new Promise<void>((resolve) => setTimeout(resolve, 20_000).unref()),
    ]);
  }
}

Why a Promise.race with a 20-second timeout? If a handler hangs — for example, waiting for a response from an external service — disconnect() will hang too. Without a timeout the process will wait forever, until the system sends SIGKILL. A 20-second timeout is a reasonable limit: if the handler did not finish in that time, disconnect is forced to complete, and NestJS continues shutting down.

The isRunning() flag in the loop lets you stop processing a batch early when the termination signal arrives, without waiting for the last message in the batch.

When to commit the offset — a common mistake

In eachBatch mode kafkajs does not commit automatically — that is your responsibility. The order matters: first the processing, then resolveOffset() and commitOffsetsIfNecessary().

A common mistake is to put resolveOffset() before the processing:

// Wrong
resolveOffset(message.offset); // offset marked as processed
await this.handle(message);    // if this throws — the message is lost

If the handler throws an exception after the offset has already been committed, the message is considered processed — and on restart it won't arrive again. The data disappears silently.

The correct order:

await this.handle(message);
resolveOffset(message.offset);
await commitOffsetsIfNecessary();

In eachMessage mode kafkajs commits on its own by default (autoCommit: true). For explicit control:

await this.consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    await this.customerService.applyUpdate(
      JSON.parse(message.value!.toString())
    );
    await this.consumer.commitOffsets([{
      topic,
      partition,
      offset: (Number(message.offset) + 1).toString(),
    }]);
  },
});

Why heavy work cannot be done directly in the handler

If the handler calls external services with retries, the total time may exceed 20 seconds. Then the timeout fires, disconnect is forced to complete, the offset is not committed — and on restart the same messages arrive. If the external call already went through, then without repeat protection you get a duplicate.

Instead of direct calls, the handler writes the intent into the database (the outbox pattern), and the real work is done by a separate process:

eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary }) => {
  for (const message of batch.messages) {
    const event: OrderConfirmedEvent = JSON.parse(message.value!.toString());
    await this.db.transaction(async (tx) => {
      const seen = await tx.query(
        'INSERT INTO processed_events(event_id, consumer_group) VALUES ($1, $2) ON CONFLICT DO NOTHING RETURNING id',
        [event.eventId, 'order-service-confirmations'],
      );
      if (seen.rowCount === 0) return;

      await tx.query(
        'INSERT INTO outbox(aggregate_id, type, payload) VALUES ($1, $2, $3)',
        [event.orderId, 'ChargePaymentRequested', JSON.stringify(event)],
      );
    });
    resolveOffset(message.offset);
    await commitOffsetsIfNecessary();
  }
},

The database transaction finishes in milliseconds — the handler always fits within the timeout. The processed_events record protects against repeats: if the same message arrives again, ON CONFLICT DO NOTHING skips it.

How to stop the producer

The producer accumulates messages in memory and sends them in batches. If the process terminates before sending, these messages are lost. An explicit producer.disconnect() call first flushes all accumulated records to the broker, then closes the connections.

@Injectable()
export class OrderEventProducer implements BeforeApplicationShutdown {
  private readonly producer: Producer;

  constructor(private readonly kafka: Kafka) {
    this.producer = kafka.producer({ idempotent: true });
  }

  async onModuleInit(): Promise<void> {
    await this.producer.connect();
  }

  async send(topic: string, messages: Message[]): Promise<void> {
    await this.producer.send({ topic, messages });
  }

  async beforeApplicationShutdown(): Promise<void> {
    await Promise.race([
      this.producer.disconnect(),
      new Promise<void>((resolve) => setTimeout(resolve, 15_000).unref()),
    ]);
  }
}

The idempotent: true option protects against duplicates on automatic retries: Kafka deduplicates records by sequence number. For critical operations an outbox is still more reliable — disconnect() may not have time to complete on an abrupt termination.

In short

  • consumer.disconnect() in beforeApplicationShutdown waits for the current eachBatch to finish and commits the offset — without it kafkajs does not wait.
  • A 20-second timeout via Promise.race is mandatory — without it a hung handler will block the shutdown forever.
  • resolveOffset() and commitOffsetsIfNecessary() are called only after a message is processed successfully, never before.
  • Heavy external calls from the handler won't fit within the timeout — move them into an outbox and a separate handler.
  • producer.disconnect() flushes accumulated records to the broker before closing; without the call, pending messages are lost.
  • Budgets and observability — how Kafka fits into the overall 60-second shutdown budget
  • HTTP drain — server.close() and how it works in parallel with the Kafka disconnect
  • Background tasks and outbox — an outbox relay with a draining-flag check
  • Kubernetes — terminationGracePeriodSeconds, preStop, readiness probes