Kafka consumers in NestJS: practical tips and trade-offs

Kafka consumers in NestJS: Building event-driven services

Event-driven architecture scales beautifully when your services react to domain events rather than polling or calling each other directly. Kafka consumers in NestJS let you build asynchronous, decoupled workflows where one service publishes an event and multiple consumers process it independently. In this post, I'll walk you through connecting Kafka to NestJS, wiring up event handlers, managing consumer groups, and ensuring reliable message processing—patterns I've used across production e-commerce, CRM, and ride-hailing platforms.

Connect with Transport.KAFKA

NestJS microservices support multiple transports out of the box. Kafka is ideal when you need durability, replay capability, and high throughput. The connection starts in your main.ts where you create a microservice instance alongside (or instead of) your HTTP listener.

Here's a practical setup:

import { NestFactory } from '@nestjs/core';
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  // Kafka microservice
  app.connectMicroservice({
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: [process.env.KAFKA_BROKER || 'localhost:9092'],
        clientId: process.env.SERVICE_NAME || 'order-service',
      },
      consumer: {
        groupId: process.env.KAFKA_GROUP_ID || 'order-service-group',
      },
    },
  });

  // Start both HTTP and microservice listeners
  await app.startAllMicroservices();
  await app.listen(3000);

  console.log('HTTP server running on port 3000');
  console.log('Kafka consumer listening...');
}

bootstrap();

The clientId identifies your service instance to the broker; the groupId groups all instances of this service into one logical consumer. If you run three replicas, they'll coordinate via Kafka to split partitions among themselves. This is the foundation for horizontal scaling.

@EventPattern('order.created') for consumers

Once Kafka is connected, you create handlers using @EventPattern decorators. These decorators bind a message topic to a method; when a message arrives on that topic, NestJS automatically deserializes it and calls your handler.

Let me show a concrete scenario: your order service publishes an order.created event, and your invoice service listens and generates an invoice.

// invoice.service.ts
import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';

export class OrderCreatedEvent {
  orderId: string;
  customerId: string;
  totalAmount: number;
  items: Array<{ productId: string; quantity: number; price: number }>;
  createdAt: Date;
}

@Controller()
export class InvoiceConsumer {
  constructor(private readonly invoiceService: InvoiceService) {}

  @EventPattern('order.created')
  async onOrderCreated(@Payload() event: OrderCreatedEvent) {
    console.log(`Processing order ${event.orderId} for customer ${event.customerId}`);
    
    try {
      await this.invoiceService.createFor(
        event.orderId,
        event.customerId,
        event.totalAmount,
        event.items,
      );
      console.log(`Invoice created successfully for order ${event.orderId}`);
    } catch (error) {
      console.error(`Failed to create invoice for order ${event.orderId}:`, error);
      throw error; // Rethrow to trigger a retry or dead-letter queue
    }
  }
}

NestJS automatically handles deserialization. The payload is converted from Kafka's binary format into your TypeScript class. If deserialization fails, the message goes to a dead-letter topic (if configured) or is retried based on your broker settings.

You can also use @EventPattern with a regular expression for topic wildcards, or define multiple patterns on one method. However, keep handlers focused—one pattern per method makes testing and debugging easier.

Use a consumer group per service

The consumer group is where horizontal scaling happens. Every service instance running with the same groupId belongs to one group. Kafka automatically assigns partitions to group members so that each partition is read by exactly one instance.

In a real-world scenario: I had an invoice service with three replicas. When an order.created` event arrived on a topic with six partitions, Kafka assigned two partitions to each replica. If a replica crashed, Kafka rebalanced and reassigned partitions to the survivors. No message was lost, and processing continued.

Set up your consumer group correctly in environment configuration:

// config/kafka.config.ts
import { Transport, MicroserviceOptions } from '@nestjs/microservices';

export const kafkaConfig: MicroserviceOptions = {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: process.env.KAFKA_BROKERS?.split(',') || ['localhost:9092'],
      clientId: `${process.env.SERVICE_NAME}-${process.env.INSTANCE_ID || 'default'}`,
      ssl: process.env.KAFKA_SSL === 'true',
      sasl: process.env.KAFKA_SASL_USERNAME
        ? {
            mechanism: 'plain',
            username: process.env.KAFKA_SASL_USERNAME,
            password: process.env.KAFKA_SASL_PASSWORD,
          }
        : undefined,
    },
    consumer: {
      groupId: process.env.KAFKA_GROUP_ID || `${process.env.SERVICE_NAME}-group`,
      sessionTimeout: 30000,
      rebalanceTimeout: 60000,
      allowAutoTopicCreation: false,
    },
  },
};

Never use the same group ID for different services. If your invoice service and your analytics service both listen to order.created, they need separate group IDs so both consume all messages. Same service, different versions? Use different group IDs during a blue-green deployment so both versions process the same events in parallel.

Commit offsets only after successful processing

Kafka tracks where each consumer has read up to using offsets. By default, NestJS commits offsets automatically, but this can lead to message loss if your handler crashes after the commit. The safer approach: commit only after your business logic succeeds.

Here's how to enable manual offset management:

// main.ts
app.connectMicroservice({
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    },
    consumer: {
      groupId: 'invoice-service-group',
      allowAutoTopicCreation: false,
    },
    run: {
      autoCommit: false, // Disable automatic commits
      autoCommitInterval: undefined,
      autoCommitThreshold: undefined,
    },
  },
});

Now your handler must commit manually. NestJS provides the full Kafka message context via the second parameter:

@EventPattern('order.created')
async onOrderCreated(
  @Payload() event: OrderCreatedEvent,
  @Ctx() context: KafkaContext,
) {
  try {
    // Process the event
    await this.invoiceService.createFor(
      event.orderId,
      event.customerId,
      event.totalAmount,
      event.items,
    );

    // Only commit if processing succeeded
    await context.getConsumer().commitOffsets([
      {
        topic: context.getTopic(),
        partition: context.getPartition(),
        offset: (Number(context.getMessage().offset) + 1).toString(),
      },
    ]);

    console.log(`Offset committed for order ${event.orderId}`);
  } catch (error) {
    console.error(`Failed to process order ${event.orderId}:`, error);
    // Do NOT commit; message will be retried
    throw error;
  }
}

Import KafkaContext from @nestjs/microservices. This pattern ensures that if your service crashes mid-processing, the offset isn't committed, and when the service restarts, it reprocesses the same message. For idempotency, store a deduplication key (like orderId) in your database so that replayed messages don't create duplicate invoices.

Common mistakes

Forgetting to set a consumer group ID: If you don't explicitly set groupId, NestJS generates a random one each time you start. Every service instance gets its own group, so all instances process all messages—no parallelism, wasted resources. Always set groupId via environment variable.

Committing before processing: Committing immediately and then failing leaves messages unprocessed. Commit only after success, or use a transactional outbox pattern where you write results and offset together.

Not handling rebalancing: When replicas join or leave, Kafka pauses all consumers for a brief rebalancing window. If your handler doesn't account for this, it may time out. Set reasonable timeouts and test with chaos engineering.

Ignoring dead-letter queues: When a message fails repeatedly, it should go somewhere you can inspect and replay it. Configure a DLQ topic and route failures there so you don't lose visibility into problems.

Blocking handlers: If your event handler is synchronous and slow, it blocks the entire consumer thread. Use async/await and consider offloading to a job queue for long-running tasks.

Key takeaways

• Connect Kafka via connectMicroservice(Transport.KAFKA) in main.ts, providing broker addresses, client ID, and consumer group ID.
• Use @EventPattern('topic.name') to bind handlers to topics; NestJS deserializes payloads automatically.
• One consumer group ID per service, enabling Kafka to coordinate partition assignment across replicas for horizontal scaling.
• Disable auto-commit and commit offsets manually after successful processing to prevent message loss on crashes.
• Always implement idempotent handlers and store deduplication keys so replayed messages don't cause side effects.
• Monitor consumer lag and rebalancing events; set appropriate session and rebalance timeouts for your use case.