ConsumerEachMessagePayload

Kind: Type

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

Part of: Microservices

Type alias to keep compatibility with

ConsumerEachMessagePayload is a compatibility type alias for the payload passed to a Kafka consumer's eachMessage handler. It provides typed access to the consumed topic, partition, message, heartbeat function, and partition-pausing controls while preserving compatibility with KafkaJS consumer APIs.

Definition

ts
EachMessagePayload

Diagram

mermaid
graph LR
  A[Kafka Consumer] --> B[eachMessage Handler]
  B --> C[ConsumerEachMessagePayload]
  C --> D[topic]
  C --> E[partition]
  C --> F[message]
  C --> G[heartbeat]
  C --> H[pause]

Usage

ts
import { ConsumerEachMessagePayload } from '@nestjs/microservices';

async function handleKafkaMessage(
  payload: ConsumerEachMessagePayload,
): Promise<void> {
  const { topic, partition, message, heartbeat } = payload;

  const value = message.value?.toString();

  console.log({
    topic,
    partition,
    offset: message.offset,
    value,
  });

  // Call during long-running message processing to keep the consumer alive.
  await heartbeat();
}

AI Coding Instructions

• Use ConsumerEachMessagePayload when typing handlers that receive KafkaJS eachMessage callback data.
• Treat message.value as nullable; check for null or use optional chaining before decoding it.
• Preserve access to heartbeat() in long-running handlers to avoid consumer session timeouts.
• Use the provided topic, partition, and message.offset values for logging, tracing, and idempotency handling.
• Keep this alias in place for KafkaJS compatibility instead of replacing it with a manually recreated payload shape.