KafkaResponseDeserializer

Kind: Class

Source: packages/microservices/deserializers/kafka-response.deserializer.ts

Part of: Microservices

KafkaResponseDeserializer converts Kafka consumer messages into Nest microservice IncomingResponse objects. It extracts the correlation ID and disposal state from Kafka headers while preserving the message value as the response payload, allowing ClientKafka to match responses to pending RPC requests.

Implements: Deserializer

Methods

Where it refuses work

• KafkaResponseDeserializer stops the work with an early return when !isUndefined(message.headers[KafkaHeaders.NEST_ERR]).
• KafkaResponseDeserializer stops the work with an early return when !isUndefined(message.headers[KafkaHeaders.NEST_IS_DISPOSED]).

Diagram

mermaid
graph LR
  A[Kafka consumer message] --> B[KafkaResponseDeserializer.deserialize]
  B --> C[Read headers]
  C --> D[Correlation ID]
  C --> E[Disposed flag]
  B --> F[Message value]
  D --> G[IncomingResponse]
  E --> G
  F --> G
  G --> H[ClientKafka RPC response handling]

Usage

ts
import {
  KafkaHeaders,
  KafkaResponseDeserializer,
} from '@nestjs/microservices';

const deserializer = new KafkaResponseDeserializer();

const incomingResponse = deserializer.deserialize({
  key: 'order.created',
  value: {
    orderId: 'order-123',
    status: 'created',
  },
  headers: {
    [KafkaHeaders.CORRELATION_ID]: 'request-abc-123',
    [KafkaHeaders.NEST_IS_DISPOSED]: false,
  },
});

console.log(incomingResponse);
// {
//   id: 'request-abc-123',
//   response: { orderId: 'order-123', status: 'created' },
//   isDisposed: false
// }

AI Coding Instructions

• Preserve the Kafka correlation ID header; it is required for matching a response with the originating RPC request.
• Keep the Kafka message value as the response payload rather than transforming its shape in this deserializer.
• Use KafkaHeaders.CORRELATION_ID and KafkaHeaders.NEST_IS_DISPOSED instead of hard-coded header names.
• Ensure response-producing Kafka consumers include the expected Nest Kafka headers when implementing custom request-response flows.

Relationships

• IMPORTS → isUndefined