Deduplicating RabbitMQ and Kafka Deliveries

Part of: Message Queue Deduplication Patterns — part of Backend Implementation & Storage Patterns

RabbitMQ and Kafka redeliver messages for different mechanical reasons — RabbitMQ on unacked or nacked channel closure, Kafka on offset-commit lag after a crash or rebalance — but the fix is the same shape in both: claim a dedup key atomically before the handler’s side effects become visible, and only acknowledge once that claim and the side effect are both durable. This is a copy-pasteable runbook, not a conceptual overview: every step below is independently verifiable.

Prerequisites: you should already understand message queue deduplication patterns — specifically the guarantee model, where the dedup key comes from, and why broker-native mechanisms (Kafka’s enable.idempotence, SQS FIFO dedup ids) do not cover consumer-side redelivery. You should also be familiar with Redis SETNX for distributed request deduplication, whose atomic-claim pattern this runbook reuses at the consumer.

---

Manual-Ack Dedup Sequence

The diagram below shows the ordering constraint that every step in this runbook depends on: the dedup store is checked before the handler runs, and the broker ack fires only after the dedup key is durably persisted — never before.

---

Problem Statement and Prerequisites

What you are implementing: a dedup guard placed between broker delivery and handler execution for two brokers with different redelivery mechanics — RabbitMQ (ack/nack on a channel) and Kafka (offset commit within a consumer group) — with the same claim-before-process invariant enforced in both.

You need:

• A running RabbitMQ node (rabbitmq-server) or Kafka broker (kafka-server-start.sh) for the verification step.
• A Redis instance reachable from the consumer, for the RabbitMQ guard.
• A Postgres (or equivalent relational) database, for the Kafka offset/key dedup table.
• Familiarity with message queue deduplication patterns, specifically the message-id-vs-business-key distinction — this runbook dedups RabbitMQ on message id and Kafka on the offset and business key together.

---

Step-by-step implementation

Step 1 — Guard RabbitMQ deliveries with a Redis SETNX check (Python, pika)

The producer must set the AMQP message_id property (RabbitMQ has no built-in id). The consumer claims that id in Redis before running the handler, and only acks after the claim and handler both succeed.

import pika
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
DEDUP_TTL_SECONDS = 86_400

def on_message(channel, method, properties, body):
    message_id = properties.message_id  # producer must set this explicitly
    if not message_id:
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        return

    dedup_key = f"rmq:dedup:{message_id}"
    claimed = r.set(dedup_key, "PROCESSING", nx=True, ex=DEDUP_TTL_SECONDS)

    if not claimed:
        # Duplicate delivery — ack immediately, do not re-run the handler
        channel.basic_ack(delivery_tag=method.delivery_tag)
        return

    try:
        process_payload(body)
        r.set(dedup_key, "DONE", ex=DEDUP_TTL_SECONDS)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        r.delete(dedup_key)  # release claim so a legitimate redelivery can retry
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
ch = connection.channel()
ch.queue_declare(queue="orders", durable=True)
ch.basic_qos(prefetch_count=10)
ch.basic_consume(queue="orders", on_message_callback=on_message, auto_ack=False)
ch.start_consuming()

The auto_ack=False flag is load-bearing — without it, RabbitMQ acks the message the instant it is delivered, before the handler or the dedup claim runs at all.

Step 2 — Mirror the RabbitMQ guard in Node.js (amqplib)

const amqp = require('amqplib');
const { createClient } = require('redis');

const DEDUP_TTL_SECONDS = 86_400;

async function main() {
  const redisClient = createClient();
  await redisClient.connect();

  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  await channel.assertQueue('orders', { durable: true });
  channel.prefetch(10);

  channel.consume('orders', async (msg) => {
    const messageId = msg.properties.messageId;
    if (!messageId) {
      channel.nack(msg, false, false);
      return;
    }

    const dedupKey = `rmq:dedup:${messageId}`;
    const claimed = await redisClient.set(dedupKey, 'PROCESSING', {
      NX: true,
      EX: DEDUP_TTL_SECONDS,
    });

    if (!claimed) {
      channel.ack(msg); // duplicate — skip handler
      return;
    }

    try {
      await processPayload(msg.content);
      await redisClient.set(dedupKey, 'DONE', { EX: DEDUP_TTL_SECONDS });
      channel.ack(msg);
    } catch (err) {
      await redisClient.del(dedupKey); // release claim for legitimate retry
      channel.nack(msg, false, true);
    }
  }, { noAck: false });
}

main().catch(console.error);

Step 3 — Guard Kafka deliveries with an offset/key dedup table (Java)

Kafka’s dedup key must combine the (topic, partition, offset) triple (cheap, exact) with the business key (catches producer republishing under a new offset). Store both columns in one row so a single query resolves either duplicate path.

@Component
public class DedupConsumer {

    private final JdbcTemplate jdbc;

    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String businessKey = extractBusinessKey(record.value()); // e.g. order_id
        long offset = record.offset();
        int partition = record.partition();

        int inserted = jdbc.update(
            "INSERT INTO kafka_dedup (topic, partition_id, kafka_offset, business_key) " +
            "VALUES (?, ?, ?, ?) ON CONFLICT (business_key) DO NOTHING",
            record.topic(), partition, offset, businessKey
        );

        if (inserted == 0) {
            // Duplicate business key — someone already claimed this event
            ack.acknowledge();
            return;
        }

        try {
            processOrder(record.value());
            ack.acknowledge(); // commit offset only after the handler succeeds
        } catch (Exception e) {
            jdbc.update("DELETE FROM kafka_dedup WHERE business_key = ?", businessKey);
            throw e; // do not ack — Kafka redelivers on the next poll
        }
    }
}

-- Dedup table: unique on business_key, indexed on (topic, partition, offset)
CREATE TABLE kafka_dedup (
    topic         TEXT   NOT NULL,
    partition_id  INT    NOT NULL,
    kafka_offset  BIGINT NOT NULL,
    business_key  TEXT   NOT NULL UNIQUE,
    consumed_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_kafka_dedup_offset ON kafka_dedup (topic, partition_id, kafka_offset);

The consumer must set enable.auto.commit=false — automatic offset commits, like RabbitMQ auto-ack, confirm the delivery before the handler and dedup write have run.

Step 4 — Mirror the Kafka guard in Go (kafka-go)

package main

import (
	"context"
	"database/sql"
	"log"

	"github.com/segmentio/kafka-go"
)

func handleMessage(ctx context.Context, db *sql.DB, r *kafka.Reader, msg kafka.Message) error {
	businessKey := extractBusinessKey(msg.Value)

	res, err := db.ExecContext(ctx,
		`INSERT INTO kafka_dedup (topic, partition_id, kafka_offset, business_key)
		 VALUES ($1, $2, $3, $4) ON CONFLICT (business_key) DO NOTHING`,
		msg.Topic, msg.Partition, msg.Offset, businessKey)
	if err != nil {
		return err
	}
	rows, _ := res.RowsAffected()
	if rows == 0 {
		return r.CommitMessages(ctx, msg) // duplicate — commit, skip handler
	}

	if err := processOrder(msg.Value); err != nil {
		db.ExecContext(ctx, `DELETE FROM kafka_dedup WHERE business_key = $1`, businessKey)
		return err // do not commit — Kafka redelivers
	}
	return r.CommitMessages(ctx, msg)
}

func main() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		GroupID: "order-processor",
		Topic:   "orders",
	})
	defer r.Close()

	db, err := sql.Open("postgres", "postgres://localhost/orders?sslmode=disable")
	if err != nil {
		log.Fatal(err)
	}

	for {
		msg, err := r.FetchMessage(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		if err := handleMessage(context.Background(), db, r, msg); err != nil {
			log.Printf("handler error, will redeliver: %v", err)
		}
	}
}

kafka-go’s FetchMessage + explicit CommitMessages mirrors the manual-ack pattern exactly — FetchMessage alone never advances the committed offset.

Step 5 — Order manual acknowledgment after the dedup write commits

Both brokers share one invariant, visible in the sequence diagram above: ack (RabbitMQ basic_ack) or offset commit (Kafka CommitMessages/Acknowledgment.acknowledge()) must fire strictly after the dedup key and the business side effect are both durable. Acking first and persisting the dedup key second reopens the exact gap this pattern exists to close — a crash between the two leaves the broker believing the message is handled while the dedup store never recorded it, and there is no redelivery left to catch the inconsistency.

Step 6 — Force redelivery and verify the guard holds

Do not trust this pattern until you have manually forced a redelivery and confirmed the handler did not run twice.

RabbitMQ: force a redelivery via nack

# Publish a test message with an explicit message_id
rabbitmqadmin publish exchange=amq.default routing_key=orders \
  payload='{"order_id":"test-1"}' properties='{"message_id":"test-msg-1"}'

# In your consumer, force a nack on the first delivery (simulate a crash before ack),
# then confirm RabbitMQ redelivers the same message_id:
rabbitmqctl list_queues name messages_unacknowledged
# Expect messages_unacknowledged to return to 0 once the second delivery is deduped and acked

Kafka: force a redelivery via offset reset

# Stop the consumer group, then rewind its committed offset to replay the last message
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-processor --topic orders \
  --reset-offsets --to-offset 0 --execute

# Restart the consumer and confirm the dedup table shows no new row for the replayed business_key
psql -c "SELECT business_key, kafka_offset, consumed_at FROM kafka_dedup WHERE business_key = 'test-1';"
# Expect exactly one row despite the replay

---

Failure scenarios and debugging

---

SRE / observability checklist

1. mq_dedup_hit_total — Counter labeled broker (rabbitmq | kafka). A sudden spike indicates a retry storm or a stuck consumer upstream, not a healthy dedup rate.
2. kafka_dedup_conflict_total — Counter incremented on every ON CONFLICT DO NOTHING no-op insert; correlate against rebalance_total to confirm conflicts track rebalances, not silent data loss.
3. consumer_manual_ack_latency_ms — Histogram measuring time from delivery to basic_ack / commitOffset; a rising p99 signals the handler or dedup store is slowing down, risking TTL/redelivery-window violations.
4. redis_unavailable_total (RabbitMQ path) / dedup_orphan_row_total (Kafka path) — either signals the dedup guard is not enforcing the claim-before-ack invariant correctly.
5. poison_message_dlq_total — Counter on messages routed to a dead-letter exchange or topic; alert on any non-zero rate.
6. Structured log fields on every dedup decision — broker, dedup_key, decision (claimed | duplicate | released), attempt_count. Index on dedup_key for incident triage across both brokers.

---

Related

• Message Queue Deduplication Patterns — parent page covering the guarantee model, broker-native vs. application dedup stores, and dedup-window sizing this runbook implements.
• Backend Implementation & Storage Patterns — parent section covering all storage-layer strategies for idempotency and exactly-once processing.
• Using Redis SETNX for Distributed Request Deduplication — the same atomic-claim pattern applied at the API gateway rather than a queue consumer.
• Idempotent Consumer Patterns for Event Streams — the fundamentals-level treatment of why consumers own the exactly-once contract.
• Deduplicating SQS and SNS Message Deliveries — the equivalent runbook for AWS-native queue and pub/sub delivery.