Event-driven architecture patterns

h Offset Management

Consumers must process events exactly once per logical unit of work. Use consumer groups for parallelism, but enforce partition-level ordering when state depends on sequence.

// consumer.ts
import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';
import { UserCreatedEventSchema } from './events';

export class EventConsumer {
  private consumer: Consumer;
  private processedIds: Set<string> = new Set();

  constructor(brokerUrl: string, groupId: string) {
    const kafka = new Kafka({ clientId: 'event-consumer', brokers: [brokerUrl] });
    this.consumer = kafka.consumer({ groupId, isolationLevel: 'read_committed' });
  }

  async init() {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'user-events', fromBeginning: false });
  }

  async run(handler: (event: UserCreatedEvent) => Promise<void>) {
    await this.consumer.run({
      eachMessage: async ({ message, partition }: EachMessagePayload) => {
        if (!message.value) return;

        const raw = message.value.toString();
        const parsed = JSON.parse(raw);
        const validated = UserCreatedEventSchema.parse(parsed);

        // Idempotency guard
        if (this.processedIds.has(validated.eventId)) {
          return;
        }

        try {
          await handler(validated);
          this.processedIds.add(validated.eventId);
          
          // Manual commit for precise offset control
          await this.consumer.commitOffsets([
            { topic: 'user-events', partition, offset: (Number(message.offset) + 1).toString() },
          ]);
        } catch (error) {
          // Route to DLQ in production; log and skip to prevent consumer stall
          console.error(`Processing failed for ${validated.eventId}:`, error);
          await this.consumer.commitOffsets([
            { topic: 'user-events', partition, offset: (Number(message.offset) + 1).toString() },
          ]);
        }
      },
    });
  }

  async disconnect() {
    await this.consumer.disconnect();
  }
}

Step 4: Architecture Decisions & Rationale

1. Idempotent Producer: Kafka’s idempotent producer prevents duplicate messages during broker retries. This is mandatory; without it, network partitions cause exactly-once semantics to break.
2. Manual Offset Commit: Auto-commit risks processing events twice on rebalance. Manual commit after successful handler execution guarantees at-least-once delivery with application-level deduplication.
3. Partition Key Strategy: Using userId as the partition key ensures all events for a single user land in the same partition, preserving causal ordering. Cross-user events can be processed in parallel.
4. Schema Validation at Ingress: Parsing and validating before business logic prevents malformed events from corrupting state. Schema evolution must be backward-compatible; breaking changes require versioned topics.
5. Dead Letter Queue (DLQ) Routing: The example logs failures and commits offsets to avoid consumer stall. Production systems must route failed messages to a DLQ topic for replay after patch deployment.

Pitfall Guide

1. Treating Events as Commands Events describe what happened. Commands describe what should happen. Routing commands through event brokers creates tight coupling and breaks idempotency. Fix: Separate command channels (HTTP/gRPC) from event channels. Publish events only after state mutation succeeds.

2. Ignoring Idempotency At-least-once delivery is the default in distributed brokers. Without idempotency keys or database-level unique constraints, duplicate processing corrupts aggregates. Fix: Store eventId in a deduplication table or use database unique indexes on event IDs.

3. Synchronous Event Handling Blocking the consumer thread for external API calls, file I/O, or long computations defeats async decoupling. Fix: Offload heavy work to worker pools, use background job queues, or split consumers into fast acknowledgment and slow processing pipelines.

4. Missing Schema Registry Unversioned payloads cause silent deserialization failures. Fix: Enforce schema validation at producer and consumer boundaries. Use a schema registry (Confluent, Apicurio) to block incompatible changes during CI/CD.

5. Poor Retry & DLQ Strategy Exponential backoff with infinite retries stalls consumers. Fix: Implement bounded retries (e.g., 3 attempts), then route to DLQ. Monitor DLQ depth as a critical alert. Never commit offsets for unhandled messages.

6. Over-Architecting Simple Workflows EDA adds operational complexity. If Service A calls Service B once per request and requires immediate feedback, REST/gRPC is superior. Fix: Reserve EDA for cross-service state propagation, audit trails, and fan-out notifications.

7. Neglecting Consumer Lag Monitoring Unmonitored lag causes stale data and delayed reactions. Fix: Track consumer_lag per partition. Alert when lag exceeds SLA thresholds. Scale consumer instances horizontally, not by increasing partition count arbitrarily.

Production Bundle

Action Checklist

• Define event schemas with strict validation and backward-compatible versioning
• Enable idempotent producers and attach unique event IDs to every message
• Implement manual offset commits after successful handler execution
• Route processing failures to a dedicated dead-letter queue topic
• Enforce partition keys for events requiring causal ordering
• Add consumer lag metrics and set alerting thresholds in observability stack
• Run chaos tests simulating broker partitions and duplicate deliveries

Decision Matrix

Configuration Template

// broker.config.ts
export const BROKER_CONFIG = {
  clientId: 'eda-service',
  brokers: [process.env.KAFKA_BROKERS || 'localhost:9092'],
  producer: {
    idempotent: true,
    transactionalId: 'eda-producer-txn-1',
    maxInFlightRequests: 5,
    retry: {
      retries: 5,
      minTimeout: 100,
      factor: 2,
    },
  },
  consumer: {
    groupId: 'eda-consumer-group',
    isolationLevel: 'read_committed',
    sessionTimeout: 30000,
    rebalanceTimeout: 60000,
    heartbeatInterval: 3000,
    maxBytesPerPartition: 1048576,
    retry: {
      retries: 3,
      minTimeout: 200,
      factor: 2,
    },
  },
  dlq: {
    topic: 'eda-dlq',
    maxRetries: 3,
    alertThreshold: 50,
  },
  schemaValidation: {
    enabled: true,
    registryUrl: process.env.SCHEMA_REGISTRY_URL,
    failOnMissingSchema: true,
  },
};

Quick Start Guide

1. Initialize Broker: Run a local Kafka instance via Docker Compose or use a managed service (Confluent Cloud, AWS MSK). Export KAFKA_BROKERS environment variable.
2. Install Dependencies: npm install kafkajs zod uuid
3. Start Producer: Import EventProducer, call init(), and publish a validated UserCreatedEvent with a UUID eventId.
4. Start Consumer: Import EventConsumer, call init(), and pass an async handler that writes to a database with a unique constraint on eventId.
5. Verify: Check consumer lag via kafka-consumer-groups.sh --describe. Confirm idempotency by republishing the same event twice; the second should be silently deduplicated.