Message Queue Backlog Explosion — When Your Queue Grows Faster Than You Consume

Production rate: 5,000 messages/second
Consumption rate: 4,800 messages/second
Net accumulation: +200 messages/second

After 1 hour:   720,000 messages
After 1 day:   17,280,000 messages
After 1 week: 120,960,000 messages

Queue latency at 1 hour: 720,000 / 4,800 = 2.5 minutes
Queue latency at 1 day: 100 minutes
Queue latency at 1 week: 7 hours

// Kubernetes HPA scaling based on queue depth (KEDA)
// KEDA: Kubernetes Event-Driven Autoscaling

// keda-scaledobject.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: order-consumer-scaledobject
spec:
  scaleTargetRef:
    name: order-consumer
  minReplicaCount: 2
  maxReplicaCount: 50
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: kafka:9092
        consumerGroup: order-processor
        topic: orders
        lagThreshold: "1000"       # Scale up when lag > 1000 messages
        offsetResetPolicy: latest

// RabbitMQ scaling with KEDA
triggers:
  - type: rabbitmq
    metadata:
      queueName: order-processing
      host: amqp://rabbitmq:5672
      queueLength: "500"    # Scale up when > 500 messages in queue

// ❌ Single-threaded consumer — bottleneck
const consumer = kafka.consumer({ groupId: 'order-processor' })
await consumer.run({
  eachMessage: async ({ message }) => {
    await processOrder(JSON.parse(message.value!.toString()))
    // Sequential: one message at a time
  }
})

// ✅ Concurrent processing within partition (for non-ordered work)
import pLimit from 'p-limit'

const limit = pLimit(10)  // 10 concurrent processors

await consumer.run({
  eachBatch: async ({ batch }) => {
    await Promise.all(
      batch.messages.map(message =>
        limit(() => processOrder(JSON.parse(message.value!.toString())))
      )
    )
    // Resolves when all 10 concurrent messages are done
  }
})

// ✅ Multiple consumer instances via consumer groups
// Run 5 instances of your consumer service
// Each instance handles different partitions

// For Kafka: number of consumers ≤ number of partitions
// If you have 10 partitions, max parallelism = 10 consumers
// Under-partitioned? Increase partition count (can't decrease later)

// Producer sends faster than consumer can handle
// Consumer should signal backpressure

class BackpressuredConsumer {
  private inFlight = 0
  private readonly maxInFlight = 100

  async startConsuming(channel: amqplib.Channel, queue: string) {
    // Tell RabbitMQ to only send N messages at a time
    // Won't send more until we ack these
    channel.prefetch(this.maxInFlight)

    channel.consume(queue, async (msg) => {
      if (!msg) return

      this.inFlight++

      try {
        await this.process(JSON.parse(msg.content.toString()))
        channel.ack(msg)
      } catch (err) {
        channel.nack(msg, false, true)  // Requeue
      } finally {
        this.inFlight--
      }
    })
  }

  // Monitor in-flight count
  getMetrics() {
    return {
      inFlight: this.inFlight,
      saturation: this.inFlight / this.maxInFlight,
    }
  }
}

// Some messages become irrelevant after a certain age
// Processing a 2-hour-old price update is worse than skipping it

// RabbitMQ: set TTL on queue
channel.assertQueue('price-updates', {
  arguments: {
    'x-message-ttl': 60_000,  // Messages expire after 60s
    'x-dead-letter-exchange': 'expired-messages',
  }
})

// Kafka: set retention per topic
// kafka-topics.sh --alter --topic price-updates
//   --config retention.ms=3600000  # 1 hour

// Application level: check message age before processing
async function processWithTTL(message: Message, maxAgeMs: number) {
  const messageAge = Date.now() - message.timestamp
  if (messageAge > maxAgeMs) {
    metrics.increment('messages.expired', { topic: message.topic })
    console.warn(`Skipping stale message: ${messageAge}ms old`)
    return  // Acknowledge and skip
  }
  await processMessage(message)
}

// When draining a large backlog, process recent messages first
// Users care about new orders more than week-old notifications

// RabbitMQ priority queue
channel.assertQueue('orders', {
  arguments: {
    'x-max-priority': 10,
  }
})

// Publish with priority (0=lowest, 10=highest)
channel.publish('', 'orders', content, {
  priority: message.timestamp > Date.now() - 60_000 ? 8 : 2,
  // Recent messages get priority 8, older ones get 2
})

// Kafka: log compaction to keep only latest per key
// Use case: inventory updates, user profile changes
// Latest value for each key is all that matters

// kafka-topics.sh --create --topic inventory
//   --config cleanup.policy=compact
//   --config min.cleanable.dirty.ratio=0.1

// Producer: use product ID as key
await producer.send({
  topic: 'inventory',
  messages: [{
    key: productId,         // Only latest message per productId kept
    value: JSON.stringify({ productId, stock: 42 }),
  }]
})

// Alert before the backlog is unrecoverable

import { Kafka } from 'kafkajs'

async function monitorKafkaLag(kafka: Kafka) {
  const admin = kafka.admin()
  await admin.connect()

  setInterval(async () => {
    const offsets = await admin.fetchTopicOffsets('orders')
    const groupOffsets = await admin.fetchOffsets({
      groupId: 'order-processor',
      topics: ['orders'],
    })

    let totalLag = 0
    for (const partition of offsets) {
      const groupOffset = groupOffsets[0].partitions
        .find(p => p.partition === partition.partition)
      const lag = parseInt(partition.offset) - parseInt(groupOffset?.offset ?? '0')
      totalLag += Math.max(0, lag)
    }

    metrics.gauge('kafka.consumer.lag', totalLag, { group: 'order-processor' })

    if (totalLag > 10_000) {
      alerting.warning(`Kafka lag: ${totalLag} messages behind`)
    }

    if (totalLag > 100_000) {
      alerting.critical(`Kafka lag critical: ${totalLag} — consumers are falling behind`)
    }
  }, 30_000)
}

// RabbitMQ queue depth via management API
async function monitorRabbitMQDepth() {
  const resp = await fetch('http://rabbitmq:15672/api/queues/%2F/orders', {
    headers: { Authorization: 'Basic ' + btoa('guest:guest') },
  })
  const queue = await resp.json()

  metrics.gauge('rabbitmq.queue.depth', queue.messages)
  metrics.gauge('rabbitmq.queue.consumers', queue.consumers)

  if (queue.messages > 50_000 && queue.consumers === 0) {
    alerting.critical('RabbitMQ queue has messages but no consumers!')
  }
}

# 1. Measure the gap
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --group order-processor \
  --describe

# 2. Scale up consumers immediately
kubectl scale deployment order-consumer --replicas=20

# 3. If messages are too old — consider skipping/archiving
# Seek consumer to recent offset to skip old messages
kafka-consumer-groups.sh --reset-offsets \
  --group order-processor \
  --topic orders \
  --to-datetime 2026-03-13T00:00:00.000 \
  --execute

# 4. Reduce message processing complexity temporarily
# Feature flag: disable non-critical side effects during drain
export DRAIN_MODE=true  # Skip analytics, emails during drain