KafkaReplyPartitionAssigner

Kind: Class

Source: packages/microservices/helpers/kafka-reply-partition-assigner.ts

Part of: Microservices

KafkaReplyPartitionAssigner is a Kafka consumer-group partition assigner used by NestJS Kafka clients for reply topics. It creates stable reply-partition assignments based on client identity, helping ensure request/reply responses are consumed by the client instance that initiated the request.

Methods

Properties

Diagram

mermaid
graph LR
  Client[Kafka client instance] --> Assigner[KafkaReplyPartitionAssigner]
  Assigner --> Protocol[protocol()]
  Assigner --> Decode[decodeMember()]
  Decode --> Previous[getPreviousAssignment()]
  Previous --> Assign[assign()]
  Assign --> Group[Kafka consumer group assignments]
  Group --> Replies[Reply topic partitions]

Usage

ts
import { Kafka } from 'kafkajs';
import { KafkaReplyPartitionAssigner } from '@nestjs/microservices/helpers/kafka-reply-partition-assigner';

const kafka = new Kafka({
  clientId: 'orders-client',
  brokers: ['localhost:9092'],
});

const replyAssigner = new KafkaReplyPartitionAssigner('orders-client');

const consumer = kafka.consumer({
  groupId: 'orders-client-replies',
  partitionAssigners: [replyAssigner],
});

await consumer.connect();
await consumer.subscribe({
  topic: 'orders.reply',
  fromBeginning: false,
});

await consumer.run({
  eachMessage: async ({ message }) => {
    console.log('Received reply:', message.value?.toString());
  },
});

AI Coding Instructions

• Configure this assigner on Kafka consumers responsible for request/reply response topics, not ordinary event-consumer groups.
• Use a stable, unique client ID for each logical Kafka client; changing it can alter reply partition ownership.
• Preserve assignment metadata compatibility when modifying protocol() or decodeMember(), since group members exchange this data during rebalances.
• Keep assign() deterministic: members and partitions should be consistently ordered to avoid unnecessary Kafka group rebalances.
• Prefer integrating through the NestJS Kafka client configuration rather than manually invoking assign() or getPreviousAssignment().

Relationships

• IMPORTS → loadPackage
• IMPORTS → isUndefined