ConsumerEachBatchPayload

Kind: Type

Source: packages/microservices/external/kafka.interface.ts

Part of: Microservices

Type alias to keep compatibility with

ConsumerEachBatchPayload is a compatibility type alias for the payload supplied to Kafka consumer eachBatch handlers. It provides access to the received batch, message offset controls, heartbeat handling, and consumer lifecycle state when processing Kafka records in batches.

Definition

ts
EachBatchPayload

Diagram

mermaid
graph LR
  Consumer[Kafka Consumer] --> Handler[eachBatch Handler]
  Handler --> Payload[ConsumerEachBatchPayload]
  Payload --> Batch[batch.messages]
  Payload --> Offset[resolveOffset]
  Payload --> Heartbeat[heartbeat]
  Payload --> State[isRunning / isStale]

Usage

ts
import type { ConsumerEachBatchPayload } from '@nestjs/microservices';

async function processKafkaBatch({
  batch,
  resolveOffset,
  heartbeat,
  isRunning,
  isStale,
}: ConsumerEachBatchPayload) {
  for (const message of batch.messages) {
    if (!isRunning() || isStale()) {
      break;
    }

    const value = message.value?.toString();
    console.log(`Processing message: ${value}`);

    // Process the message before marking its offset as resolved.
    resolveOffset(message.offset);

    // Keep the Kafka consumer session alive during long-running work.
    await heartbeat();
  }
}

AI Coding Instructions

• Use this type for Kafka eachBatch callbacks to preserve compatibility with the underlying Kafka consumer payload.
• Check isRunning() and isStale() while iterating through large batches to avoid processing invalid or revoked work.
• Call resolveOffset() only after a message has been processed successfully.
• Invoke heartbeat() during long-running batch processing to prevent consumer session timeouts.
• Treat batch.messages as ordered Kafka records and handle message values that may be null.