Design a Distributed Message Queue: Avoid the 3 AM Thread Pool Exhaustion

The 3 AM page that wakes you up: 'Payment service latency spike — messages stuck in queue.' You SSH in, see 10,000 unacknowledged messages, and the consumer thread pool is completely deadlocked. This isn't a Kafka bug. It's a design failure you introduced. Most engineers think a distributed message queue is just 'Kafka with a topic.' They're wrong. The hard part isn't the queue — it's the guarantees: exactly-once semantics, ordering across partitions, and handling a broker crash without losing a single message. By the end of this, you'll design a queue that survives a node failure, a network partition, and a consumer that takes 30 seconds to process a message — without dropping data or deadlocking.

Core Architecture: Partitions Are Your Scalability Unit

Before you write a single line, understand this: a distributed message queue is a partitioned commit log. Each partition is an ordered, immutable sequence of messages. Producers append to partitions, consumers read from them. The magic is that partitions can live on different brokers, giving you horizontal scaling. But here's the trap: ordering is only guaranteed within a partition. If you need global ordering, you need a single partition — which kills throughput. Choose wisely. In production, we use a key-based partitioning scheme: all messages for the same order ID go to the same partition. That preserves order per order, while allowing parallel consumption across orders. Never hash by random — you'll get hot partitions.

// io.thecodeforge — System Design tutorial

// Producer: partition by order ID
ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
    "order-events",
    order.getOrderId(), // key -> same partition
    order.toEvent()
);
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // Retry with exponential backoff, log to dead letter queue
        deadLetterQueue.send(record);
    }
});

// Consumer: assign partitions to threads
// Each thread owns a set of partitions
for (TopicPartition partition : assignment) {
    executor.submit(() -> {
        while (true) {
            ConsumerRecords<String, OrderEvent> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, OrderEvent> record : records) {
                processOrder(record.value());
            }
            consumer.commitSync(); // commit after batch
        }
    });
}

Durability: The Commit Log and Replication Factor

A queue that loses messages is a liability. Durability comes from the commit log — every message is written to disk on the broker before being acknowledged. But disk writes are slow. So we batch them. In Kafka, the producer can set acks=all and linger.ms=5 to batch 5ms worth of messages into a single disk write. That gives you durability with 10x throughput. Replication factor (RF) of 3 means three brokers have a copy. If one dies, the others take over. But RF=3 means three times the storage and network traffic. For non-critical logs, RF=2 is fine. For payment events, RF=3 minimum. Never set RF=1 in production — I've seen a single disk failure wipe out a day of orders.

// io.thecodeforge — System Design tutorial

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("acks", "all"); // wait for all in-sync replicas
props.put("retries", Integer.MAX_VALUE); // infinite retries on transient errors
props.put("enable.idempotence", true); // exactly-once semantics
props.put("linger.ms", 5); // batch up to 5ms
props.put("batch.size", 65536); // 64KB batches
props.put("compression.type", "snappy"); // reduce network bandwidth

KafkaProducer<String, OrderEvent> producer = new KafkaProducer<>(props);

Consumer Groups and Rebalancing: The Silent Killer

Consumer groups let you scale consumption: multiple consumers split partitions among themselves. But when a consumer joins or leaves, a rebalance triggers — all consumers stop, revoke partitions, and reassign. During rebalance, no messages are processed. In a 100-partition topic, rebalance can take 30 seconds. That's 30 seconds of piling messages. The fix: use cooperative rebalancing (Kafka 2.4+) which reassigns partitions incrementally. Also set session.timeout.ms high enough (45s) to avoid false rebalances on GC pauses. And never use static group membership unless you want manual partition assignment — it's a maintenance nightmare.

// io.thecodeforge — System Design tutorial

Properties consumerProps = new Properties();
consumerProps.put("group.id", "order-processor");
consumerProps.put("session.timeout.ms", 45000); // 45s before considered dead
consumerProps.put("heartbeat.interval.ms", 15000); // send heartbeat every 15s
consumerProps.put("partition.assignment.strategy", 
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); // incremental rebalance
consumerProps.put("max.poll.interval.ms", 300000); // 5min max between polls
consumerProps.put("max.poll.records", 100); // process 100 records per poll

KafkaConsumer<String, OrderEvent> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Arrays.asList("order-events"));

Exactly-Once Semantics: The Holy Grail and Its Cost

At-least-once is easy: retry on failure. But duplicates happen. Exactly-once requires idempotent producers and transactional consumers. The producer assigns a unique ID to each message; the broker deduplicates. The consumer reads messages and writes results in a transaction — if it crashes, the transaction is rolled back, and the message is re-delivered. But transactions add latency (2x-3x). For payment processing, it's worth it. For logging, it's overkill. Use exactly-once only when duplicates cause financial loss. Otherwise, use idempotent consumers (dedup by message ID) — cheaper and simpler.

// io.thecodeforge — System Design tutorial

// Consumer with idempotent processing
consumer.subscribe(Arrays.asList("order-events"));
while (true) {
    ConsumerRecords<String, OrderEvent> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, OrderEvent> record : records) {
        String messageId = record.key(); // unique per message
        if (processedIds.contains(messageId)) {
            continue; // dedup
        }
        processOrder(record.value());
        processedIds.add(messageId);
    }
    consumer.commitSync();
}

// For exactly-once with transactions:
producer.initTransactions();
producer.beginTransaction();
for (ConsumerRecord<String, OrderEvent> record : records) {
    producer.send(new ProducerRecord<>("processed-orders", record.key(), record.value()));
}
producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

Backpressure and Flow Control: Don't Let the Queue Eat Your Memory

If producers outpace consumers, the queue grows unbounded. Brokers run out of disk. Consumers get overwhelmed. The fix: backpressure. Producers should block when the queue is full. Kafka doesn't have native backpressure — you implement it with max.in.flight.requests.per.connection (set to 1 for strict ordering) and a callback that blocks. Better: use a rate limiter on the consumer side. If consumer processing time spikes, slow down polling. Also set retention limits: delete messages older than 7 days or when topic size exceeds 100GB. Never let a topic grow forever — I've seen a 2TB topic that took 3 days to clean up.

// io.thecodeforge — System Design tutorial

// Rate-limited consumer
RateLimiter limiter = RateLimiter.create(100.0); // 100 permits per second
while (true) {
    limiter.acquire(); // blocks if rate exceeded
    ConsumerRecords<String, OrderEvent> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, OrderEvent> record : records) {
        processOrder(record.value());
    }
    consumer.commitSync();
}

// Broker-side retention
// Set on topic creation or via CLI:
// bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name order-events --alter --add-config retention.ms=604800000,retention.bytes=107374182400

Monitoring and Alerting: What to Watch Before It Burns

You can't fix what you don't measure. Monitor these metrics: consumer lag (difference between latest offset and consumer offset), request rate, error rate, and disk usage. Use Prometheus + Grafana. Alert on consumer lag > 1000 for more than 5 minutes — that means consumers are falling behind. Also alert on under-replicated partitions (URP) — that means a broker is down and data might be at risk. And monitor GC pauses: if young GC takes > 200ms, your heap is too small. I've seen a 10-second GC pause cause a consumer group rebalance that took down a whole microservice.

// io.thecodeforge — System Design tutorial

# Prometheus alert rule for consumer lag
groups:
  - name: kafka_alerts
    rules:
      - alert: ConsumerLagHigh
        expr: kafka_consumer_lag > 1000
        for: 5m
        annotations:
          summary: "Consumer lag > 1000 for topic {{ $labels.topic }}"
      - alert: UnderReplicatedPartitions
        expr: kafka_server_ReplicaManager_UnderReplicatedPartitions > 0
        for: 1m
        annotations:
          summary: "Under-replicated partitions detected"

# Check lag via CLI
# bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group order-processor --describe

When Not to Use a Distributed Message Queue

Distributed message queues add operational complexity: brokers to manage, partitions to tune, rebalances to handle. If you have a single server and low throughput (< 100 msg/s), use an in-memory queue (e.g., Disruptor) or a simple database table with a status column. If you need strict FIFO across all messages, a single-partition queue limits throughput — consider a database with ordered IDs. And if your messages are tiny and latency-critical (< 1ms), a distributed queue's network overhead kills you. Use a shared memory queue instead. I've seen teams adopt Kafka for a 10 msg/s internal logging pipeline — they spent more time managing Kafka than building features. Don't be that team.