Consumer Acknowledgements #

In the RabbitMQ broker ecosystem, upstream delivery guarantees alone are not enough. After a message is successfully published by the producer and stored by the broker, the final part of the delivery guarantee lies entirely on the downstream side: Consumer Acknowledgements.

Consumer Acknowledgements is an active communication protocol from consumer applications back to the broker declaring task completion status. Through this mechanism, RabbitMQ can know when a message is safe to delete from permanent storage, when it should be redelivered to another consumer due to connection failures, or when it should be isolated to another channel. Errors in configuring and managing consumer confirmations can cause operational disasters in production, from silent message loss, total queue stalls, to RAM memory leaks that cripple the broker. This article thoroughly discusses consumer confirmation semantics, the close relationship between ACKs and Prefetch QoS quotas, the memory leak dangers of unconfirmed messages, safe transaction ordering, and Go implementations for robust confirmation handling.

The Position of Consumer Confirmation in the Message Lifecycle #

Consumer confirmation is the closing operation determining the end of a message’s journey in the broker queue.

flowchart TD
    Queue["Queue (Message Status: Ready)"] -->|"basic.deliver"| Consumer["Consumer Application"]
    Consumer -->|"Message Changes Status: Unacknowledged"| Logic["Business Logic Execution & DB Commit"]
    
    Logic --> Result{"Success?"}
    
    Result -->|Yes| Ack["Send basic.ack"]
    Result -->|"No (Temporary Error)"| NackRequeue["Send basic.nack (requeue=true)"]
    Result -->|"No (Permanent Error)"| NackDLX["Send basic.nack (requeue=false)"]
    
    Ack --> Delete["Message Deleted from Queue RAM/Disk"]
    NackRequeue --> QueueRequeue["Message Requeued (Back to Queue)"]
    NackDLX --> DLX["Message Diverted to Dead Letter Exchange (DLX)"]

When the broker sends a message to a consumer, the message transitions to the Unacknowledged status. At this point, the broker tracks message ownership based on the consumer’s TCP connection. The broker won’t release that message until it receives an ACK/NACK signal back, or detects the TCP connection dropped.

Confirmation Semantics: Manual ACK, NACK, and REJECT #

The AMQP 0-9-1 protocol provides specific confirmation instructions for consumers to communicate a message’s final fate to the broker:

1. basic.ack (Positive Acknowledgement) #

This command is sent when the consumer application has successfully completed message processing and committed the business database transaction without issues.

  • Implication: The RabbitMQ broker immediately deletes that message from the queue (RAM and disk), and restores the consumer’s prefetch quota to receive the next message.

2. basic.nack (Negative Acknowledgement) #

A rejection command introduced by RabbitMQ to signal process failures. Unlike regular rejects, basic.nack supports batch rejection of messages in groups.

  • requeue = true Option: Returns the message to the original queue at the front position for another attempt. Suitable for temporary errors (database connection drops).
  • requeue = false Option: Deletes the message from the original queue. If the queue has a Dead Letter Exchange (DLX), the message is diverted to that DLX; if not, the message is discarded forever. Suitable for permanent errors (corrupted JSON format).

3. basic.reject (Single Rejection) #

The standard AMQP 0-9-1 rejection instruction. Behavior is the same as basic.nack, but only limited to processing one single message per network call, without batching support.

State Tracking and Erlang BEAM Memory Management #

At the RabbitMQ broker’s internal level, Unacknowledged status management uses very efficient Erlang fast memory data structures.

1. ETS (Erlang Term Storage) Tables per Channel #

Every consumer channel (rabbit_channel) maintains an internal ETS memory table recording the list of sequence tags of in-flight messages being processed by the consumer. This table stores the relationship between local delivery_tags, physical message IDs in the queue, and remaining prefetch quota allocations.

  • Fast Lookup: When a consumer sends basic.ack(102), the broker performs a microsecond-latency ETS lookup to find the original message reference in the queue.
  • Cleanup: After the entry is found, the data in the ETS table is immediately cleaned and memory space is freed right away.

2. Off-Heap Memory Cleanup (refc Binaries) #

Large message body payloads are allocated by the Erlang VM outside the queue heap area (off-heap storage) using a reference counting system.

  • When a message is pushed to a consumer, the reference counter is incremented.
  • While the message is in the Unacknowledged status, the reference counter must not be zero, forcing the operating system to keep holding that message’s data bytes in the server’s physical RAM.
  • Only after the consumer sends an ACK (which drops the reference counter to zero) can the Erlang Garbage Collector (GC) asynchronously delete the message bytes from the server’s physical RAM memory. Failing to send ACKs automatically blocks this GC process, slowly but surely triggering broker RAM memory leaks.

ACK Correlation with Prefetch QoS and Memory Leaks #

One of the areas most often triggering operational failures in production is the interaction between Consumer Acknowledgements and Prefetch QoS (Quality of Service).

How Does Prefetch Limit Unacknowledged Messages? #

Prefetch QoS limits the maximum number of Unacknowledged status messages allowed to flow to one consumer channel. If we set prefetch = 100, the broker only sends a maximum of 100 messages. The 101st message delivery is suspended until the consumer sends at least one ACK to free the quota.

The Danger of Forgetting to Send ACKs (Memory Leak & Queue Stall) #

If we make an error in writing consumer application code (e.g., forgetting to write the ACK call in one of the if/else error-handling branches, or the processing goroutine deadlocks before calling the ACK):

  1. Queue Stall: The consumer receives the first 100 messages and processes them. However, because no ACK is sent back to the broker, the consumer’s prefetch quota stays full forever. The RabbitMQ broker stops sending new messages to that consumer. The consumer application appears stalled with no new activity, while new messages keep piling up in the queue.
  2. Broker RAM Memory Leak: RabbitMQ must keep holding the message body payloads of those 100 Unacknowledged messages in its RAM memory in case the consumer disconnects. If thousands of consumer channels experience this mass ACK-forgetting problem, broker RAM memory balloons past the high memory watermark, triggering memory alarms that freeze the entire RabbitMQ cluster activity (flow control).

Therefore, monitoring the Unacknowledged message count metric through Prometheus/Grafana dashboards is mandatory in production.

Safe Execution Order: Commit the Database First, Then Send the ACK #

When designing transactional consumers, the code writing order between main database execution and ACK sending is the determinant of data safety.

1. Fatal Order: ACK Before Database Commit (Anti-Pattern) #

Some developers write code that sends the ACK first to free up the queue, then processes the slow local database transaction.

flowchart LR
    A["Receive Message"] --> B["Send ACK to Broker"] --> C["Execute Database"] --> D["Commit Fails! (Data Lost)"]
  • Risk: If right after the ACK is sent (and the broker deletes the message from its queue), our local database server crashes or loses power while trying to commit business data, that message is already physically gone from the broker and can’t be recovered. This causes permanent data loss.

2. Safe Order: Commit the Database First, Then Send the ACK (Solution) #

The correct and safe code writing order must prioritize the local database first:

flowchart LR
    A["Receive Message"] --> B["Execute Database"] --> C["DB Commit Success"] --> D["Send ACK to Broker"]
  • Advantage: If the database server fails to commit the transaction or the consumer application crashes before the final step, the database transaction is automatically rolled back and the RabbitMQ broker (which hasn’t received the ACK) detects the lost consumer connection and returns the message to the queue (requeue). Data is safe from loss risk.

Channel Recovery and Re-consumption Strategies #

When a network disruption momentarily cuts the TCP connection, all Unacknowledged messages on that disconnected channel are automatically returned by the broker to the original queue.

Disciplined Consumer Recovery Steps: #

  1. Closure Detection: Our consumer application must listen for channel closure notifications (NotifyClose).
  2. Open a New Connection: Re-negotiate the TCP connection and open a new channel.
  3. Reset Prefetch QoS: Re-set the prefetch QoS limit on that new channel.
  4. Re-register the Consumer: Call the basic.consume command again to re-register our consumer.
  5. Local Status Cleanup: Empty or re-align the internal message tracking cache in consumer memory so it doesn’t collide with new delivery_tags that the broker will restart from number 1.

Go Code Implementation: Transactional Consumer with Manual ACK and Graceful Shutdown #

Here is a complete Go language implementation example for a robust consumer. This code registers a consumer with manual ACK, sets the QoS prefetch limit, evaluates processing error types to determine requeue choices, and handles system signals for graceful shutdown to ensure in-flight messages finish being ACKed safely before the connection is dropped.

package main

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

const (
	amqpURL   = "amqp://guest:***@localhost:5672/"
	queueName = "payment-transactions-queue"
)

type PaymentPayload struct {
	PaymentID string  `json:"payment_id"`
	UserID    string  `json:"user_id"`
	Amount    float64 `json:"amount"`
}

func main() {
	// 1. Open a TCP Connection to RabbitMQ
	conn, err := amqp.Dial(amqpURL)
	if err != nil {
		log.Fatalf("Gagal terhubung ke RabbitMQ: %s", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Gagal membuka channel: %s", err)
	}
	defer ch.Close()

	// 2. Configure the QoS Prefetch (Prefetch Limit)
	// We limit the quota to a maximum of 30 unacknowledged messages per channel
	err = ch.Qos(
		30,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	if err != nil {
		log.Fatalf("Gagal menyetel QoS Prefetch: %s", err)
	}

	// 3. Declare a Durable Queue
	_, err = ch.QueueDeclare(
		queueName,
		true,  // durable
		false, // auto-delete
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal mendeklarasikan antrean: %s", err)
	}

	// 4. Register the Consumer (autoAck = false)
	msgs, err := ch.Consume(
		queueName,
		"",    // consumer tag
		false, // autoAck: false (manual ACK for the At-Least-Once guarantee!)
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan konsumen: %s", err)
	}

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	log.Println("[*] Konsumen aktif. Menunggu pesan...")

	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case d, ok := <-msgs:
				if !ok {
					log.Println("[!] Channel pesan ditutup.")
					return
				}

				// Transactional message processing
				processPaymentMessage(d)
			}
		}
	}()

	// Wait for the OS termination signal for Graceful Shutdown
	<-sigChan
	log.Println("[*] Menerima sinyal terminasi. Memulai graceful shutdown...")
	
	// Cancel the context so the goroutine stops receiving new messages
	cancel()
	
	// Give tolerance time for in-flight messages to send their final ACKs
	time.Sleep(1500 * time.Millisecond)
	log.Println("[*] Shutdown aman diselesaikan.")
}

func processPaymentMessage(d amqp.Delivery) {
	// Define a defer recovery to handle unexpected panics in the processing code
	defer func() {
		if r := recover(); r != nil {
			log.Printf("[PANIC] Terjadi kepanikan saat memproses pesan: %v. Mengirimkan NACK requeue...", r)
			// Send NACK requeue = true so the message is retried on another instance
			d.Nack(false, true)
		}
	}()

	var payment PaymentPayload
	
	// 5. Evaluate the Payload Format (Permanent Error)
	err := json.Unmarshal(d.Body, &payment)
	if err != nil {
		log.Printf("[ERROR] Payload rusak untuk ID Pesan %s: %s. Menolak pesan permanen...", d.MessageId, err)
		// Send NACK with requeue = false so the message is discarded to the Dead Letter Queue (DLQ)
		d.Nack(false, false)
		return
	}

	// 6. Evaluate Business Logic Validity (Permanent Error)
	if payment.Amount <= 0 {
		log.Printf("[ERROR] Nilai pembayaran tidak valid (Amount: %.2f) untuk ID %s. Mengabaikan pesan...", payment.Amount, payment.PaymentID)
		// Send NACK requeue = false so it doesn't clog the main queue
		d.Nack(false, false)
		return
	}

	// 7. Execute the Main Database Transaction Logic
	dbSuccess := savePaymentToDatabase(payment)
	if !dbSuccess {
		log.Printf("[WARN] Kegagalan database sementara saat memproses %s. Melakukan requeue...", payment.PaymentID)
		// Send NACK with requeue = true so it's retried later after the database recovers
		d.Nack(false, true)
		return
	}

	// 8. Send the Manual ACK after a SUCCESSFUL database commit
	err = d.Ack(false)
	if err != nil {
		log.Printf("[ERROR] Gagal mengirimkan ACK ke broker: %s", err)
	} else {
		log.Printf("✓ Transaksi %s sukses terproses dan di-ACK.", payment.PaymentID)
	}
}

func savePaymentToDatabase(payment PaymentPayload) bool {
	// Simulate a database function
	return true
}

Anti-Patterns vs Practical Solutions in Production #

Avoid the following fatal mistakes in message receipt confirmation management:

Anti-Pattern: Sending ACKs from Asynchronous Goroutines Without Channel Scope Synchronization #

Receiving messages in the main consumer loop, immediately throwing the processing task and the d.Ack(false) function call into a new goroutine declared asynchronously via the go process(d) command, without limiting the number of parallel goroutines running.

Why is this wrong? #

This pattern triggers two major dangers:

  1. Channel Scope Exception: The RabbitMQ driver library is not designed to be thread-safe if several goroutines try to call ACK/NACK operations on the same channel simultaneously at random. This can trigger the AMQP 406 PRECONDITION_FAILED protocol error that force-closes the channel connection.
  2. Loss of Backpressure Control: Because new goroutines are created without limits, we unconsciously bypass the prefetch QoS quota restriction. The consumer keeps sucking messages from the socket and hoarding them in internal application RAM memory, triggering OutOfMemory crashes.
  • Solution: Limit our consumer goroutine concurrency using a measured Worker Pool Pattern. Make sure ACK/NACK signal delivery returns to the main channel-managing goroutine sequentially, or use internal mutex/channel synchronization to secure ACK calls on the AMQP channel object.

Summary #

  • Consumer Reliability Control — Consumer Acknowledgements determine when messages are safe to delete from broker queues (basic.ack) or must be resent (basic.nack).
  • Manual ACK Mechanism — A mandatory best practice for transactional production systems. Setting Auto-Ack (autoAck = true) is highly vulnerable to message loss when applications crash.
  • The Impact of Forgetting ACKs — Causes RAM memory leaks on the RabbitMQ broker and exhausts QoS prefetch quotas, freezing queue data flow (queue stalls).
  • Transaction Order Rules — Make sure application code always completes business logic and commits local database transactions first, then sends the ACK instruction to the broker.
  • Rejection with Requeue — Use requeue = true for temporary infrastructure errors, and requeue = false (diverting to the DLX) for permanent data format errors.
  • Safe Goroutine Concurrency — Avoid random ACK calls across goroutines without channel synchronization to prevent PRECONDITION_FAILED exceptions that disconnect connections.

← Previous: Publisher confirms   Next: Requeue vs Drop →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact