Section 1: The Monolithic Bottleneck and the Event-Driven Shift

When scaling an enterprise platform, the limitations of traditional monolithic architectures quickly become apparent. In a typical REST-based synchronous system, services call each other directly over HTTP. While this is simple to implement initially, it introduces tight coupling. If Service A relies on Service B, and Service B relies on Service C, a failure in Service C propagates upstream, causing a cascading failure that can bring down the entire system. Additionally, synchronous calls tie up connection pools and block threads, leading to high resource utilization and latency spikes under load.

Synchronous HTTP communication creates an architectural dependency graph that is highly fragile. When traffic surges, client request threads are blocked waiting for network I/O from downstream systems. This blocks the runtime thread pool (such as Node.js's libuv threads or Tomcat's HTTP threads), starving the system of capacity to handle incoming requests. Even if a service is op\x65rational, it can be forced offline because its connection pools to databases or dependencies are exhausted by slow HTTP requests. Additionally, retry policies with backoff can worsen this problem by creating retry storms, where failing services are bombarded with traffic, preventing recovery.

To solve this, modern architectures shift toward event-driven systems. By communicating asynchronously through an event broker, services become decoupled. Instead of calling a service directly, a service publishes an event (e.g., "Order Created"). Interested services subscribe to this event and process it independently. If a downstream service is temporarily offline, the event broker stores the message, ensuring zero data loss and enabling the service to catch up once it recovers.

In this deep dive, we will walk through building an enterprise-grade event-driven architecture using NestJS for microservice development, Apache Kafka as our highly scalable event broker, and Redis for distributed caching, rate limiting, and idempotency checks.

---

Section 2: Kafka Architecture: Topics, Partitions, and Consumer Groups

To design a resilient event pipeline, we must understand how Kafka handles messages. Unlike traditional message queues like RabbitMQ, Kafka is an append-only commit log. Messages are written to topics, which are divided into partitions.

\x60\x60\x60text Topic: order-events +-------------------------+ | Partition 0: [0][1][2] | ---> Consumer A (Group: billing-service) +-------------------------+ | Partition 1: [0][1][2] | ---> Consumer B (Group: billing-service) +-------------------------+ | Partition 2: [0][1][2] | ---> Consumer C (Group: billing-service) +-------------------------+ \x60\x60\x60

Key Architectural Concepts:

1. Partitions for Scalability: Partitions are the unit of scalability in Kafka. Each partition can be hosted on a different broker, allowing a topic to handle throughput far beyond the capacity of a single server.
2. Order Guarantees: Kafka guarantees message order within a partition, but not across the entire topic. To maintain ordering (e.g., ensuring "Payment Processed" is processed after "Order Created"), we must route related events to the same partition using a Partition Key (such as the \x60orderId\x60).
3. Consumer Groups: Multiple instances of a service join a consumer group. Kafka automatically distributes partitions among the active consumers in a group. If one instance crashes, Kafka triggers a rebalance, reassigning its partitions to the remaining healthy instances.
4. Commit Offsets and Lag Tracking: Consumers track their progress by committing offsets back to a special Kafka topic (\x60__consumer_offsets\x60). Consumer lag is the difference between the latest offset written by the producer and the latest offset committed by the consumer. Minimizing consumer lag is a primary op\x65rational objective.
5. Static Membership to Minimize Rebalances: Under standard configurations, when a consumer node restarts, Kafka triggers a "stop-the-world" rebalance, pausing all partition consumption in the group. By configuring static membership via the \x60group.instance.id\x60 property, we allow a consumer to restart and reclaim its assigned partitions without triggering a rebalance, provided it reconnects within the configured session timeout window.
6. Broker Disk Persistence and Log Segments: Kafka writes events to append-only disk files called log segments. By sequentializing I/O op\x65rations and using the OS page cache alongside zero-copy system calls (like \x60sendfile\x60), Kafka achieves high performance, matching network throughput limits.

---

Section 3: Implementing the NestJS Kafka Microservice

Let's write a concrete implementation. We will build an Order Service that publishes events, and a Payment Service that consumes them.

1. Order Service (Producer) Configuration

First, we define our Kafka client configuration in the Order Service's NestJS module:

\x60\x60\x60typescript // order.module.ts import { Module } from '@nestjs/common'; import { ClientsModule, Transport } from '@nestjs/microservices'; import { OrderController } from './order.controller'; import { OrderService } from './order.service';

@Module({ imports: [ ClientsModule.register([ { name: 'KAFKA_SERVICE', transport: Transport.KAFKA, options: { client: { clientId: 'order-service', brokers: ['localhost:9092'], connectionTimeout: 10000, requestTimeout: 25000, ssl: false, // Set to true in production with appropriate certificates }, consumer: { groupId: 'order-consumer-group', allowAutoTopicCreation: true, }, }, }, ]), ], controllers: [OrderController], providers: [OrderService], }) export class OrderModule {} \x60\x60\x60

2. Publishing Events from the Controller

Next, we inject the Kafka client and emit an event when an order is created. We will implement key-based routing to ensure all events for a given order land on the same partition:

\x60\x60\x60typescript // order.controller.ts import { Controller, Post, Body, Inject, OnModuleInit } from '@nestjs/common'; import { ClientKafka } from '@nestjs/microservices'; import { OrderService } from './order.service';

@Controller('orders') export class OrderController implements OnModuleInit { constructor( @Inject('KAFKA_SERVICE') private readonly kafkaClient: ClientKafka, private readonly orderService: OrderService, ) {}

async onModuleInit() { this.kafkaClient.subscribeToResponseOf('order-created-topic'); await this.kafkaClient.connect(); }

@Post() async createOrder(@Body() orderDto: { userId: string; amount: number; items: string[] }) { const newOrder = await this.orderService.saveToDatabase(orderDto);

// Emit the event to Kafka, using the order ID as the partition key
this.kafkaClient.emit('order-created-topic', {
  key: newOrder.id,
  value: JSON.stringify({
    orderId: newOrder.id,
    userId: newOrder.userId,
    amount: newOrder.amount,
    items: newOrder.items,
    timestamp: new Date().toISOString(),
  }),
  headers: {
    'correlation-id': 'corr-' + Math.random().toString(36).substr(2, 9),
    'x-app-version': '1.0.0',
  }
});

return { status: 'PENDING_PAYMENT', orderId: newOrder.id };

} } \x60\x60\x60

3. Payment Service (Consumer) Implementation

Now, let's create our consumer microservice that listens to the \x60order-created-topic\x60 and processes payments.

First, configure the microservice bootstrap file:

\x60\x60\x60typescript // main.ts (Payment Microservice) import { NestFactory } from '@nestjs/core'; import { MicroserviceOptions, Transport } from '@nestjs/microservices'; import { PaymentModule } from './payment.module';

async function bootstrap() { const app = await NestFactory.createMicroservice<MicroserviceOptions>(PaymentModule, { transport: Transport.KAFKA, options: { client: { brokers: ['localhost:9092'], clientId: 'payment-service', }, consumer: { groupId: 'payment-consumer-group', sessionTimeout: 30000, heartbeatInterval: 10000, metadataMaxAge: 300000, }, }, }); await app.listen(); console.log('Payment Microservice is listening...'); } bootstrap(); \x60\x60\x60

Inside the Controller, we handle the incoming message:

\x60\x60\x60typescript // payment.controller.ts import { Controller } from '@nestjs/common'; import { MessagePattern, Payload, Ctx, KafkaContext } from '@nestjs/microservices'; import { PaymentService } from './payment.service';

@Controller() export class PaymentController { constructor(private readonly paymentService: PaymentService) {}

@MessagePattern('order-created-topic') async handleOrderCreated( @Payload() message: any, @Ctx() context: KafkaContext, ) { const originalMessage = context.getMessage(); const partition = context.getPartition(); const topic = context.getTopic(); const correlationId = originalMessage.headers ? originalMessage.headers['correlation-id'] : 'N/A';

console.log(\x60[CorrelationID: \x24{correlationId}] Processing message from partition \x24{partition} of topic \x24{topic}\x60);

const eventData = typeof message === 'string' ? JSON.parse(message) : message;
await this.paymentService.processPayment(eventData.orderId, eventData.amount);

} } \x60\x60\x60

---

Section 4: Ensuring Idempotency and Distributed Locking with Redis

In an event-driven system, networks can fail. If a consumer successfully processes an event but crashes before committing the offset back to Kafka, Kafka will redeliver the event. This is known as At-Least-Once delivery.

To prevent duplicate processing (e.g., charging a customer twice for the same order), consumers must be Idempotent. We enforce this using Redis.

When an event is received, we store the unique event key in Redis with a Time-To-Live (TTL). If the key already exists, we skip processing.

\x60\x60\x60typescript // payment.service.ts import { Injectable, Inject, Logger } from '@nestjs/common'; import { CACHE_MANAGER } from '@nestjs/cache-manager'; import { Cache } from 'cache-manager';

@Injectable() export class PaymentService { private readonly logger = new Logger(PaymentService.name);

constructor( @Inject(CACHE_MANAGER) private readonly cacheManager: Cache, ) {}

async processPayment(orderId: string, amount: number): Potential<void> { const idempotencyKey = \x60idempotency:payment:\x24{orderId}\x60;

// Check if event was already processed
const isProcessed = await this.cacheManager.get<string>(idempotencyKey);
if (isProcessed) {
  this.logger.warn(\x60Duplicate event received for Order ID: \x24{orderId}. Skipping...\x60);
  return;
}

// Acquire lock and save state in a single atomic transaction
await this.cacheManager.set(idempotencyKey, 'processing', 3600); // 1-hour TTL

try {
  // Execute payment charging transaction
  await this.executeTransaction(orderId, amount);
  
  // Update state to completed
  await this.cacheManager.set(idempotencyKey, 'completed', 86400); // 24-hour TTL
} catch (error) {
  // Release lock on failure to allow retry
  await this.cacheManager.del(idempotencyKey);
  throw error;
}

}

private async executeTransaction(orderId: string, amount: number): Potential<void> { this.logger.log(\x60Charging \x24\x24{amount} for Order \x24{orderId}\x60); // API Call to Stripe or internal gateway } } \x60\x60\x60

Atomic Lock Acquisition Using Redis Lua Scripts

To prevent race conditions where two threads process the same event concurrently (e.g. if partition assignment is modified and two nodes receive the same event momentarily), we use a Redis Lua script to acquire a lock atomically. This ensures that only one worker can write to Redis and proceed.

\x60\x60\x60typescript // redis-lock.service.ts import { Injectable } from '@nestjs/common'; import { InjectRedis } from '@liaoliaots/nestjs-redis'; import Redis from 'ioredis';

@Injectable() export class RedisLockService { constructor(@InjectRedis() private readonly redis: Redis) {}

// Lua script ensuring check-and-set atomic op\x65rations private readonly acquireScript = \x60 if redis.call('get', KEYS[1]) == false then redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2]) return 1 else return 0 end \x60;

async acquireLock(key: string, value: string, ttlMs: number): Potential<boolean> { const result = await this.redis.eval( this.acquireScript, 1, key, value, ttlMs.toString() ); return result === 1; }

async releaseLock(key: string, value: string): Potential<boolean> { const releaseScript = \x60 if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end \x60; const result = await this.redis.eval(releaseScript, 1, key, value); return result === 1; } } \x60\x60\x60

---

Section 5: Resiliency and Error Handling: Retry Strategies & Dead Letter Queues (DLQ)

In production, downstream systems fail. Databases deadlock, APIs time out, and networks drop packets. We need a resilient resiliency strategy.

We implement a three-tier retry pipeline:

1. Immediate In-Memory Retry: For transient errors, retry up to 3 times with an exponential backoff.
2. Retry Topic (Delayed Queue): If the error persists, push the event to a \x60payment-retry\x60 topic. A separate consumer processes this topic with a delay (e.g., 5 minutes) to give the external dependency time to recover.
3. Dead Letter Queue (DLQ): If all retries fail, push the event to \x60payment-dlq\x60 for manual inspection and alerts.

\x60\x60\x60typescript // error.interceptor.ts import { Injectable, NestInterceptor, ExecutionContext, CallHandler, Inject } from '@nestjs/common'; import { Observable, throwError } from 'rxjs'; import { catchError } from 'rxjs/op\x65rators'; import { ClientKafka } from '@nestjs/microservices';

@Injectable() export class KafkaErrorInterceptor implements NestInterceptor { constructor(@Inject('KAFKA_SERVICE') private readonly kafkaClient: ClientKafka) {}

intercept(context: ExecutionContext, next: CallHandler): Observable<any> { const ctx = context.switchToRpc().getContext(); const originalMessage = ctx.getMessage(); const topic = ctx.getTopic();

return next.handle().pipe(
  catchError((error) => {
    const headers = originalMessage.headers || {};
    const retryCount = parseInt(headers['x-retry-count'] || '0', 10);
    
    if (retryCount < 3) {
      const nextRetryTopic = \x60\x24{topic}-retry\x60;
      headers['x-retry-count'] = (retryCount + 1).toString();
      
      console.warn(\x60Retrying message on topic \x24{nextRetryTopic}. Attempt: \x24{retryCount + 1}\x60);
      
      this.kafkaClient.emit(nextRetryTopic, {
        key: originalMessage.key,
        value: originalMessage.value,
        headers,
      });
    } else {
      const dlqTopic = \x60\x24{topic}-dlq\x60;
      console.error(\x60Max retries reached. Routing to DLQ: \x24{dlqTopic}. Error: \x24{error.message}\x60);
      
      this.kafkaClient.emit(dlqTopic, {
        key: originalMessage.key,
        value: JSON.stringify({
          originalPayload: originalMessage.value,
          failedAt: new Date().toISOString(),
          errorMessage: error.message,
          stack: error.stack,
        }),
      });
    }
    
    return throwError(() => error);
  }),
);

} } \x60\x60\x60

---

Section 6: Production Monitoring: Prometheus Metrics and Dashboards

Observability is vital for managing event-driven systems. We use Prometheus to track consumer lag, processing rates, and system errors.

NestJS Prometheus Service Implementation

\x60\x60\x60typescript // metrics.service.ts import { Injectable } from '@nestjs/common'; import { Counter, Histogram, Registry } from 'prom-client';

@Injectable() export class MetricsService { private readonly registry: Registry; public readonly eventCounter: Counter<string>; public readonly processingDuration: Histogram<string>; public readonly dlqCounter: Counter<string>;

constructor() { this.registry = new Registry();

this.eventCounter = new Counter({
  name: 'kafka_events_processed_total',
  help: 'Total number of Kafka events processed',
  labelNames: ['topic', 'status'],
  registers: [this.registry],
});

this.processingDuration = new Histogram({
  name: 'kafka_event_processing_duration_seconds',
  help: 'Time spent processing Kafka events in seconds',
  labelNames: ['topic'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 5, 10],
  registers: [this.registry],
});

this.dlqCounter = new Counter({
  name: 'kafka_events_routed_to_dlq_total',
  help: 'Total events routed to the Dead Letter Queue',
  labelNames: ['topic', 'reason'],
  registers: [this.registry],
});

}

getMetrics(): Potential<string> { return this.registry.metrics(); } } \x60\x60\x60

Prometheus Alertmanager Rules

Save these alert rules in \x60/etc/prometheus/alert_rules.yml\x60 to trigger notifications when op\x65rational performance issues arise:

\x60\x60\x60yaml groups:

• name: KafkaAlertRules rules:

  • alert: KafkaConsumerLagHigh expr: sum(kafka_consumergroup_lag) by (consumergroup, topic) > 500 for: 2m labels: severity: warning annotations: summary: "High consumer group lag detected" description: "Consumer group {{ \x24labels.consumergroup }} on topic {{ \x24labels.topic }} is lagging by {{ \x24value }} messages."

  • alert: KafkaDLQWriteDetected expr: increase(kafka_events_routed_to_dlq_total[1m]) > 0 for: 0m labels: severity: critical annotations: summary: "Event routed to Dead Letter Queue (DLQ)" description: "An event on topic {{ \x24labels.topic }} was routed to the DLQ due to: {{ \x24labels.reason }}." \x60\x60\x60

Grafana Dashboard Configuration JSON

This JSON template configures a Grafana panel for real-time monitoring of consumer status and processing latency:

\x60\x60\x60json { "annotations": { "list": [] }, "editable": true, "fiscalYearStartMonth": 0, "graphTooltip": 0, "id": 101, "links": [], "liveNow": false, "panels": [ { "collapsed": false, "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, "id": 1, "title": "Kafka Event Processing Rate", "type": "timeseries", "targets": [ { "datasource": { "type": "prometheus", "uid": "prometheus" }, "editorMode": "code", "expr": "rate(kafka_events_processed_total[1m])", "legendFormat": "{{topic}} - {{status}}", "range": true, "refId": "A" } ] }, { "collapsed": false, "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, "id": 2, "title": "95th Percentile Event Processing Latency", "type": "timeseries", "targets": [ { "datasource": { "type": "prometheus", "uid": "prometheus" }, "editorMode": "code", "expr": "histogram_quantile(0.95, sum(rate(kafka_event_processing_duration_seconds_bucket[5m])) by (le, topic))", "legendFormat": "{{topic}}", "range": true, "refId": "A" } ] } ], "refresh": "5s", "schemaVersion": 38, "style": "dark", "tags": ["kafka", "nestjs", "observability"], "time": { "from": "now-1h", "to": "now" }, "timepicker": {}, "timezone": "", "title": "Kafka & Microservices Dashboard", "version": 1 } \x60\x60\x60

---

Section 7: Kafka Cluster Topology Configuration

Below is a three-broker local cluster setup running in KRaft mode (no Zookeeper dependency) alongside a Redis cluster node, configured for optimal development parity:

\x60\x60\x60yaml

docker-compose.yml

version: '3.8'

services: kafka-1: image: confluentinc/cp-kafka:7.4.0 container_name: kafka-1 ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT' KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-1:29092,PLAINTEXT_HOST://localhost:9092' KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_PROCESS_ROLES: 'broker,controller' KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093' KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092' KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs' CLUSTER_ID: 'MkU3OEVBRjFGM0U0NEFCN0'

kafka-2: image: confluentinc/cp-kafka:7.4.0 container_name: kafka-2 ports: - "9094:9094" environment: KAFKA_NODE_ID: 2 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT' KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-2:29092,PLAINTEXT_HOST://localhost:9094' KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_PROCESS_ROLES: 'broker,controller' KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093' KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9094' KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs' CLUSTER_ID: 'MkU3OEVBRjFGM0U0NEFCN0'

kafka-3: image: confluentinc/cp-kafka:7.4.0 container_name: kafka-3 ports: - "9096:9096" environment: KAFKA_NODE_ID: 3 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT' KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka-3:29092,PLAINTEXT_HOST://localhost:9096' KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_PROCESS_ROLES: 'broker,controller' KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka-1:29093,2@kafka-2:29093,3@kafka-3:29093' KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9096' KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs' CLUSTER_ID: 'MkU3OEVBRjFGM0U0NEFCN0'

redis: image: redis:7.0-alpine container_name: redis-cache ports: - "6379:6379" command: redis-server --save 60 1 --loglevel warning volumes: - redis_data:/data

volumes: redis_data: \x60\x60\x60

---

Section 8: Testing Strategy: Integration Testing with Testcontainers

This integration test verifies that the system correctly routes messages, enforces idempotency, and processes events. It uses Testcontainers to orchestrate a transient environment during validation sweeps:

\x60\x60\x60typescript // payment.integration.spec.ts import { Test, TestingModule } from '@nestjs/testing'; import { INestMicroservice } from '@nestjs/common'; import { MicroserviceOptions, Transport } from '@nestjs/microservices'; import { PaymentModule } from './payment.module'; import { PaymentService } from './payment.service'; import { KafkaContainer } from '@testcontainers/kafka'; import { RedisContainer } from '@testcontainers/redis'; import Redis from 'ioredis';

describe('Payment Service Event Pipeline Integration Test', () => { let app: INestMicroservice; let kafkaContainer: KafkaContainer; let redisContainer: RedisContainer; let redisClient: Redis; let paymentService: PaymentService;

beforeAll(async () => { // Start temporary Docker containers kafkaContainer = await new KafkaContainer().withExposedPorts(9093).start(); redisContainer = await new RedisContainer().start();

const bootstrapBrokers = [\x60\x24{kafkaContainer.getHost()}:\x24{kafkaContainer.getMappedPort(9093)}\x60];

redisClient = new Redis({
  host: redisContainer.getHost(),
  port: redisContainer.getMappedPort(6379),
});

const moduleFixture: TestingModule = await Test.createTestingModule({
  imports: [PaymentModule],
})
  .overrideProvider('REDIS_CLIENT')
  .useValue(redisClient)
  .compile();

paymentService = moduleFixture.get<PaymentService>(PaymentService);

app = moduleFixture.createNestMicroservice<MicroserviceOptions>({
  transport: Transport.KAFKA,
  options: {
    client: { brokers: bootstrapBrokers },
    consumer: { groupId: 'test-payment-consumer-group' },
  },
});

await app.listen();

}, 120000);

aft\x65rAll(async () => { await app.close(); await redisClient.quit(); await kafkaContainer.stop(); await redisContainer.stop(); });

it('should process payment transaction only once for a given orderId', async () => { const orderId = 'order-99999-idempotence'; const amount = 250.00;

const paymentSpy = jest.spyOn(paymentService as any, 'executeTransaction');

// Run two consecutive calls simulating event replication
await paymentService.processPayment(orderId, amount);
await paymentService.processPayment(orderId, amount);

// Verify transaction was called once
expect(paymentSpy).toHaveBeenCalledTimes(1);

const savedState = await redisClient.get(\x60idempotency:payment:\x24{orderId}\x60);
expect(savedState).toBe('completed');

}); }); \x60\x60\x60

---

Section 9: Conclusion

By moving from synchronous REST calls to an event-driven architecture with NestJS, Kafka, and Redis, we achieved decoupled scalability, high resilience, and low latency. Using partitions ensures logical order, and Redis provides idempotency, creating a production-ready event processing pipeline.