kafka-patterns

name	kafka-patterns
description	Topic design, partition strategies, consumer group patterns, exactly-once processing, and dead letter queue handling.

Kafka Patterns

Event streaming patterns for Apache Kafka in distributed systems.

Topic Design

# Topic naming convention: <domain>.<entity>.<event-type>
# Examples:
#   orders.order.created
#   payments.payment.completed
#   inventory.stock.updated

# Topic configuration
topics:
  orders.order.created:
    partitions: 12          # Match expected consumer parallelism
    replication-factor: 3   # Survive 2 broker failures
    retention.ms: 604800000 # 7 days
    cleanup.policy: delete

  orders.order.changelog:
    partitions: 12
    replication-factor: 3
    retention.ms: -1        # Infinite retention (compacted)
    cleanup.policy: compact # Keep latest value per key
    min.compaction.lag.ms: 3600000  # 1h before compacting

Producer Patterns

import { Kafka, Partitioners, CompressionTypes } from 'kafkajs'

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: process.env.KAFKA_BROKERS!.split(','),
})

const producer = kafka.producer({
  idempotent: true,                                    // Exactly-once producer
  maxInFlightRequests: 5,                              // Max parallel requests
  createPartitioner: Partitioners.DefaultPartitioner,
})

await producer.connect()

// Key-based partitioning: same key always goes to same partition (ordering)
async function publishOrderEvent(order: Order, eventType: string): Promise<void> {
  await producer.send({
    topic: `orders.order.${eventType}`,
    compression: CompressionTypes.LZ4,
    messages: [{
      key: order.,                    
      : .({
        : crypto.(),   
        eventType,
        :  ().(),
        : order,
      }),
      : {
        : ,
        : ,
        : order.,
      },
    }],
  })
}


  (): <> {
   producer.({
    : [{
      : ,
      : events.( ({
        : e.,
        : .(e),
      })),
    }],
  })
}

Consumer Group Patterns

const consumer = kafka.consumer({
  groupId: 'payment-processor',     // Consumer group: shared topic consumption
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
  maxBytesPerPartition: 1048576,    // 1MB per partition per fetch
  retry: { retries: 5 },
})

await consumer.connect()
await consumer.subscribe({
  topics: ['orders.order.created'],
  fromBeginning: false,              // Start from latest offset
})

await consumer.run({
  autoCommit: false,                 // Manual commit for exactly-once
  eachBatchAutoResolve: false,

  eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary, heartbeat }) => {
    for (const message of batch.messages) {
      try {
        const event = JSON.parse(message.value!.toString())

        // Idempotency check: skip already processed events
        if (await isAlreadyProcessed(event.eventId)) {
          (message.)
          
        }

         (event.)
         (event.)

        (message.)
         ()
         ()
      }  (err) {
        .(, err)
        
         (message, err  )
        (message.)
      }
    }
  },
})

Dead Letter Queue (DLQ)

const DLQ_TOPIC = 'orders.order.created.dlq'

async function sendToDeadLetterQueue(
  originalMessage: KafkaMessage,
  error: Error
): Promise<void> {
  await producer.send({
    topic: DLQ_TOPIC,
    messages: [{
      key: originalMessage.key,
      value: originalMessage.value,
      headers: {
        ...originalMessage.headers,
        'dlq-reason': error.message,
        'dlq-timestamp': new Date().toISOString(),
        'dlq-original-topic': 'orders.order.created',
        'dlq-retry-count': '0',
      },
    }],
  })
}

// DLQ consumer: retry or alert
async function processDLQ(): Promise<void> {
  const dlqConsumer = kafka.consumer({ groupId: 'dlq-processor' })
  await dlqConsumer.subscribe({ topics: [DLQ_TOPIC] })

   dlqConsumer.({
    :  ({ message }) => {
       retryCount = (
        message.?.[]?.() ?? 
      )

       (retryCount >= ) {
        
         ({
          : ,
          : message.?.(),
          : message.?.[]?.(),
          : retryCount,
        })
        
      }

      
       {
         event = .(message.!.())
         (event.)
      }  (err) {
        
         producer.({
          : ,
          : [{
            : message.,
            : message.,
            : {
              ...message.,
              : (retryCount + ),
            },
          }],
        })
      }
    },
  })
}

Partition Strategy

// Custom partitioner: route by region for data locality
const regionalPartitioner = () => ({
  partition: ({ topic, partitionMetadata, message }) => {
    const region = message.headers?.['region']?.toString() ?? 'default'
    const regionMap: Record<string, number> = {
      'us-east': 0, 'us-west': 1,
      'eu-west': 2, 'eu-east': 3,
      'ap-southeast': 4,
    }
    const partition = regionMap[region]
    if (partition !== undefined && partition < partitionMetadata.length) {
      return partition
    }
    // Fallback: hash the key
    const numPartitions = partitionMetadata.length
    const hash = murmurHash(message.key?.toString() ?? '')
    return Math.abs(hash) % numPartitions
  }
})

Checklist

• Idempotent producer enabled (exactly-once semantics)
• Key-based partitioning for ordered processing per entity
• Manual offset commit (not auto-commit) for at-least-once guarantee
• Dead Letter Queue for failed messages (max 3 retries then alert)
• Consumer idempotency check before processing (eventId dedup)
• Heartbeat during long-running batch processing
• Replication factor >= 3 for production topics
• Monitoring: consumer lag, throughput, error rate per consumer group

Anti-Patterns

• Auto-commit offsets: message loss if consumer crashes before processing
• Single partition topics: no parallelism, bottleneck
• Unbounded retry: infinite retry loop blocks partition processing
• Large messages (>1MB): use claim-check pattern (store in S3, send reference)
• Skipping idempotency: duplicate processing on consumer restart
• Global ordering requirement: use single partition only when truly needed