Message Queue Components — The Auto-ack Data Loss Gotcha

package io.thecodeforge;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.client.ConfirmListener;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * Production-grade producer with durable messaging and publisher confirms.
 * io.thecodeforge naming conventions throughout.
 *
 * Two critical settings:
 *   1. durable=true on the queue declaration — survives broker restart
 *   2. PERSISTENT_TEXT_PLAIN delivery mode — message flushed to disk before broker ACKs
 *
 * Publisher confirms (channel.confirmSelect()) ensure we know when the broker
 * has actually received and persisted the message, not just buffered it.
 */
public class OrderProducer {
    private static final String QUEUE_NAME = "order_processing";

    // Track outstanding unconfirmed publishes: sequenceNumber -> message payload
    // In production this would trigger a retry or fallback path on nack
    private final ConcurrentNavigableMap<Long, String> outstandingConfirms =
        new ConcurrentSkipListMap<>();

    public void publishOrder(String customerId, double amount) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("forge_user");
        factory.setPassword("forge_password");

        // One Connection per process — creating a new Connection per publish
        // adds 100-200ms of TCP handshake overhead and exhausts file descriptors at scale.
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // durable=true: this queue survives a broker restart
            // exclusive=false: allow multiple consumers
            // autoDelete=false: queue persists when the last consumer disconnects
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            // Enable publisher confirms — the broker will ack or nack each message
            // so we know it was received and persisted, not just buffered in memory
            channel.confirmSelect();

            // Wire up confirm listeners before publishing
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) {
                    if (multiple) {
                        outstandingConfirms.headMap(deliveryTag + 1).clear();
                    } else {
                        outstandingConfirms.remove(deliveryTag);
                    }
                    System.out.printf(" [Forge] Broker confirmed delivery for tag: %d%n", deliveryTag);
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) {
                    // Broker rejected the message — trigger your retry or fallback path
                    String nacked = outstandingConfirms.get(deliveryTag);
                    System.err.printf(" [Forge] Broker NACK'd message, triggering fallback: %s%n", nacked);
                    // In production: write to outbox table for later retry
                }
            });

            // Generate a unique event ID that consumers use for idempotency checks
            String eventId = UUID.randomUUID().toString();
            String payload = String.format(
                "{\"event_id\":\"%s\", \"customer_id\":\"%s\", \"amount\":%.2f, \"currency\":\"USD\"}",
                eventId, customerId, amount
            );

            // Track this publish before sending so the confirm listener can correlate
            long sequenceNumber = channel.getNextPublishSeqNo();
            outstandingConfirms.put(sequenceNumber, payload);

            // PERSISTENT_TEXT_PLAIN = Delivery Mode 2 + Content-Type text/plain
            // Mode 2 tells the broker to write to disk before acknowledging
            // Without this, a broker crash between receipt and flush loses the message
            channel.basicPublish(
                "",               // default exchange — routes directly to named queue
                QUEUE_NAME,       // routing key for default exchange = queue name
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                payload.getBytes(StandardCharsets.UTF_8)
            );

            System.out.printf(" [Forge] Published order event %s for customer %s%n",
                eventId, customerId);

            // Wait for broker confirmation before returning to caller
            // In high-throughput scenarios, use async confirms and batch publishing instead
            if (!channel.waitForConfirms(5000)) {
                throw new RuntimeException("Broker did not confirm message within 5 seconds");
            }
        }
    }
}

version: '3.8'

services:
  thecodeforge-broker:
    image: rabbitmq:3.13-management
    container_name: forge-rabbitmq
    ports:
      - "5672:5672"   # AMQP protocol — application connections
      - "15672:15672" # Management UI — use for debugging queue depth and consumer health
    environment:
      RABBITMQ_DEFAULT_USER: forge_user
      RABBITMQ_DEFAULT_PASS: forge_password
      # Reduce heartbeat to detect dropped connections faster (default is 60s)
      RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: "-rabbitmq_management load_definitions /etc/rabbitmq/definitions.json"
    volumes:
      # Named volume ensures messages survive docker-compose down and up
      # Without this, every restart wipes the queue — dangerous for DLQ testing
      - rabbitmq_data:/var/lib/rabbitmq
      # Pre-load exchange and queue definitions on startup for dev-prod parity
      - ./rabbitmq-definitions.json:/etc/rabbitmq/definitions.json:ro
    healthcheck:
      # Wait for the broker to be ready before starting dependent services
      test: ["CMD", "rabbitmq-diagnostics", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 20s

volumes:
  rabbitmq_data:
    # Explicit driver declaration makes the volume's purpose visible to future engineers
    driver: local

package io.thecodeforge;

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * Production-grade consumer with:
 *   1. Manual acknowledgement — only ack after all side effects succeed
 *   2. Idempotency check — safe to receive the same message twice
 *   3. Prefetch=1 — fair dispatch across all consumer instances
 *   4. DLQ routing on permanent failure — no infinite poison message loops
 */
public class OrderConsumer {
    private static final String QUEUE_NAME = "order_processing";
    private static final String DB_URL     = "jdbc:postgresql://localhost:5432/forge";
    private static final String DB_USER    = "forge_user";
    private static final String DB_PASS    = "forge_password";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("forge_user");
        factory.setPassword("forge_password");

        // Enable automatic connection recovery — reconnects after network interruptions
        // without requiring consumer code to restart or re-register callbacks
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000); // retry every 5 seconds

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // Declare the queue with the same parameters as the producer
        // If the queue already exists with different params, this throws — catch it early
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // CRITICAL: prefetch=1 prevents one consumer from hoarding all messages
        // The broker will not deliver a new message until the previous one is acked
        // This forces round-robin distribution across all running consumer instances
        channel.basicQos(1);

        System.out.println(" [Forge] Consumer waiting for messages on: " + QUEUE_NAME);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String body = new String(delivery.getBody(), StandardCharsets.UTF_8);
            long deliveryTag = delivery.getEnvelope().getDeliveryTag();
            String eventId   = extractEventId(body);

            System.out.printf(" [Forge] Received event %s%n", eventId);

            try {
                // Step 1: Idempotency check — has this event been processed before?
                // Required because at-least-once delivery means duplicates WILL arrive
                if (isAlreadyProcessed(eventId)) {
                    System.out.printf(" [Forge] Duplicate event %s — acking without processing%n", eventId);
                    channel.basicAck(deliveryTag, false);
                    return;
                }

                // Step 2: Execute business logic — all of this must succeed or fail atomically
                processOrder(body);

                // Step 3: Mark as processed in the idempotency table
                // Do this inside the same DB transaction as processOrder() in production
                markAsProcessed(eventId);

                // Step 4: Only ack AFTER all side effects are committed to durable storage
                // If we crash between processOrder() and basicAck(), the broker redelivers
                // The idempotency check in Step 1 handles that redelivery safely
                channel.basicAck(deliveryTag, false);
                System.out.printf(" [Forge] Successfully processed and acked event %s%n", eventId);

            } catch (Exception error) {
                System.err.printf(" [Forge] Failed to process event %s: %s%n",
                    eventId, error.getMessage());

                // requeue=false: send to DLQ instead of requeueing
                // requeue=true would create an infinite loop for permanent failures
                // The DLQ preserves the message for manual inspection and replay
                channel.basicNack(deliveryTag, false, false);
            }
        };

        // autoAck=false: broker keeps message unacked until we call basicAck or basicNack
        // This is the single most important setting — never set this to true in production
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag ->
            System.out.println(" [Forge] Consumer cancelled: " + consumerTag)
        );
    }

    private static void processOrder(String json) throws Exception {
        // In production: parse JSON, call payment gateway, write to DB in one transaction
        System.out.printf(" [Forge] Processing order payload: %s%n", json);
        Thread.sleep(50); // simulate processing time
    }

    private static boolean isAlreadyProcessed(String eventId) throws Exception {
        try (var conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS);
             PreparedStatement ps = conn.prepareStatement(
                 "SELECT 1 FROM processed_events WHERE event_id = ? LIMIT 1")) {
            ps.setString(1, eventId);
            ResultSet rs = ps.executeQuery();
            return rs.next();
        }
    }

    private static void markAsProcessed(String eventId) throws Exception {
        try (var conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASS);
             PreparedStatement ps = conn.prepareStatement(
                 "INSERT INTO processed_events (event_id, processed_at) VALUES (?, NOW()) ON CONFLICT DO NOTHING")) {
            ps.setString(1, eventId);
            ps.executeUpdate();
        }
    }

    private static String extractEventId(String json) {
        // In production use Jackson or Gson for JSON parsing
        // This simple extraction is for illustration clarity only
        int start = json.indexOf('"', json.indexOf("event_id") + 10) + 1;
        int end   = json.indexOf('"', start);
        return json.substring(start, end);
    }
}

-- io.thecodeforge: Schema for auditing and replaying DLQ extractions
-- This table provides the context that the message broker cannot preserve:
-- which queue failed, why it failed, how many times it was attempted,
-- and whether it has been replayed or resolved.

CREATE TABLE dlq_archive (
    id                  BIGSERIAL       PRIMARY KEY,
    original_event_id   UUID            NOT NULL,
    source_queue        VARCHAR(255)    NOT NULL,  -- which queue the message came from
    payload             JSONB           NOT NULL,  -- full message body for replay
    failure_reason      TEXT            NOT NULL,  -- exception message or error code
    retry_count         INT             NOT NULL DEFAULT 0,
    first_failed_at     TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
    last_failed_at      TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
    resolved_at         TIMESTAMPTZ,               -- null until an engineer resolves it
    resolution_notes    TEXT                       -- what was done to fix or discard
);

-- Fast lookup by event_id for idempotency checks and deduplication during replay
CREATE UNIQUE INDEX idx_dlq_event_id
    ON dlq_archive(original_event_id);

-- For ops dashboards: filter by source queue and unresolved status
CREATE INDEX idx_dlq_source_unresolved
    ON dlq_archive(source_queue, resolved_at)
    WHERE resolved_at IS NULL;

-- Query: find all unresolved failures for the order_processing queue in the last 24 hours
SELECT
    original_event_id,
    failure_reason,
    retry_count,
    first_failed_at,
    payload->>'customer_id'  AS customer_id,
    payload->>'amount'       AS amount
FROM dlq_archive
WHERE source_queue  = 'order_processing'
  AND resolved_at   IS NULL
  AND first_failed_at >= NOW() - INTERVAL '24 hours'
ORDER BY first_failed_at DESC;

-- Replay query: select fixable messages to republish to the original queue
-- After fixing the consumer bug, republish these payloads with new event_ids
SELECT original_event_id, payload
FROM dlq_archive
WHERE source_queue    = 'order_processing'
  AND failure_reason  LIKE '%connection timeout%'  -- transient errors worth replaying
  AND resolved_at     IS NULL;