Test Message Queue Consumers in Go Without a Broker

A queue consumer is a loop with three decisions in it: acknowledge a message that was handled, drop one that can never succeed, and put back one that might succeed later. Those decisions are the part worth testing, and they do not need Kafka, RabbitMQ or SQS. They need a way to hand messages to the loop and see what it did with each.

This article shows that seam on Go 1.25.6: a Delivery type that stands for "a message with a way to settle it", a consumer loop written against a channel, and a recorder that turns the broker's reply into a slice of strings the test can compare. The pattern works with any broker client, because every one of them reduces to the same three calls.

The seam

package main

import (
	"context"
	"errors"
	"sync"
)

// Delivery is what a broker client hands to a handler: a body plus a way to acknowledge it.
type Delivery struct {
	ID   string
	Body []byte
	Ack  func()
	Nack func(requeue bool)
}

var ErrPermanent = errors.New("permanent failure")

type Handler func(ctx context.Context, body []byte) error

// Consume reads deliveries until the channel closes or ctx ends.
// Success acks, a permanent error drops the message, anything else requeues it.
func Consume(ctx context.Context, in <-chan Delivery, h Handler) {
	seen := map[string]bool{}
	var mu sync.Mutex
	for {
		select {
		case <-ctx.Done():
			return
		case d, ok := <-in:
			if !ok {
				return
			}
			mu.Lock()
			dup := seen[d.ID]
			mu.Unlock()
			if dup { // redelivery of something already handled
				d.Ack()
				continue
			}
			switch err := h(ctx, d.Body); {
			case err == nil:
				mu.Lock()
				seen[d.ID] = true
				mu.Unlock()
				d.Ack()
			case errors.Is(err, ErrPermanent):
				d.Nack(false)
			default:
				d.Nack(true)
			}
		}
	}
}

func main() {}

The boundary is the Delivery struct. A real adapter would build one from the broker's message: for RabbitMQ, Ack calls d.Ack(false) and Nack calls d.Nack(false, requeue); for SQS, Ack deletes the message and Nack changes its visibility timeout; for Kafka, Ack marks the offset and Nack is a dead letter publish. The consumer never imports a broker package, so it compiles and tests anywhere.

Consume has one input, a receive-only channel. It returns when the context ends or when the channel is closed, which is how the tests stop it. In production the adapter feeds the channel from the broker's subscription and closes it on shutdown, and a graceful stop is then the same code path as the test's. The shutdown side of that story is in testing graceful shutdown.

The handler returns an error, and the consumer classifies it with errors.Is(err, ErrPermanent). A handler that wraps the sentinel with fmt.Errorf("parse: %w", ErrPermanent) is still permanent, so handlers can add context. The error testing article covers why errors.Is is the right comparison.

The fake broker

package main

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"testing"
)

// recorder is a fake broker: it collects what the consumer decided for each delivery.
type recorder struct{ events []string }

func (r *recorder) delivery(id, body string) Delivery {
	return Delivery{
		ID: id, Body: []byte(body),
		Ack:  func() { r.events = append(r.events, id+":ack") },
		Nack: func(requeue bool) { r.events = append(r.events, fmt.Sprintf("%s:nack(requeue=%v)", id, requeue)) },
	}
}

func run(t *testing.T, h Handler, ds ...func(*recorder) Delivery) []string {
	t.Helper()
	r := &recorder{}
	in := make(chan Delivery, len(ds))
	for _, d := range ds {
		in <- d(r)
	}
	close(in) // a closed channel ends Consume, so the test needs no sleep and no timeout
	Consume(t.Context(), in, h)
	return r.events
}

func msg(id, body string) func(*recorder) Delivery {
	return func(r *recorder) Delivery { return r.delivery(id, body) }
}

func TestOutcomes(t *testing.T) {
	h := func(_ context.Context, b []byte) error {
		switch string(b) {
		case "bad":
			return fmt.Errorf("parse: %w", ErrPermanent)
		case "flaky":
			return errors.New("database unavailable")
		}
		return nil
	}
	got := run(t, h, msg("1", "ok"), msg("2", "bad"), msg("3", "flaky"))
	t.Logf("%v", got)
	want := []string{"1:ack", "2:nack(requeue=false)", "3:nack(requeue=true)"}
	if !slices.Equal(got, want) {
		t.Fatalf("got %v want %v", got, want)
	}
}

func TestRedeliveryHandledOnce(t *testing.T) {
	calls := 0
	h := func(_ context.Context, b []byte) error { calls++; return nil }
	got := run(t, h, msg("7", "x"), msg("7", "x"), msg("7", "x"))
	t.Logf("handler calls=%d events=%v", calls, got)
	if calls != 1 {
		t.Fatalf("handler ran %d times", calls)
	}
}

go test -count=1 -v . 2>&1 | grep -vE '^(=== RUN|=== PAUSE|=== CONT|PASS|ok)' | sed -E 's/ \([0-9.]+s\)//; s#^ +[a-z_]+\.go:[0-9]+: #    #'

    [1:ack 2:nack(requeue=false) 3:nack(requeue=true)]
--- PASS: TestOutcomes
    handler calls=1 events=[7:ack 7:ack 7:ack]
--- PASS: TestRedeliveryHandledOnce

The recorder is the entire fake. Each Delivery it builds has Ack and Nack functions that append a string to one slice: 1:ack, or 2:nack(requeue=false). Using strings for the log makes the assertion a single slices.Equal against a literal, and a failure prints both slices side by side, with the order visible. It is the same idea as recording calls in a hand-written fake, but the thing recorded is the decision rather than the call.

run loads all the deliveries into a buffered channel, closes it, and calls Consume on the test goroutine. Because the channel is closed after the last message, Consume returns after processing it. There is no goroutine to wait for, no WaitGroup, no sleep and no timeout. That is the main design point: a consumer that ends when its input ends is testable by calling it.

TestOutcomes feeds three messages and checks three outcomes. The first is handled and acknowledged. The second returns an error wrapping ErrPermanent, which the consumer settles with Nack(false); a broker with a dead letter queue routes it there. The third returns a plain error and is settled with Nack(true), which puts it back. The recorded log is [1:ack 2:nack(requeue=false) 3:nack(requeue=true)], in message order.

At least once delivery

Most brokers deliver at least once. A message can arrive twice because an acknowledgement was lost, a consumer restarted, or a visibility timeout expired during a slow handler. A consumer that is not idempotent processes the order twice.

TestRedeliveryHandledOnce sends the same ID three times. The consumer remembers the IDs it handled and acknowledges a repeat without calling the handler, so the log shows handler calls=1 and [7:ack 7:ack 7:ack]. All three are acknowledged, because an unacknowledged duplicate would be delivered again forever.

The map in this lab is the simplest possible deduplication, and it has the limits you would expect: it lives in memory, so a restart forgets it, and it grows without bound. In a real system the record belongs in the same database transaction as the effect of the handler, with a unique key on the message ID, so "handled" and "recorded as handled" cannot disagree. The test for that version needs a database; the transaction rollback article shows how to give each test a clean one. What this lab keeps is the contract: a duplicate is acknowledged and never reaches the handler.

What to add next

The two tests are a base. A production consumer needs a few more, each one cheap with this setup.

• Context cancellation. Cancel the context before sending and assert that Consume returns without reading the channel. Use an unbuffered channel and a deadline so a regression fails rather than hangs.
• Handler panics. Decide whether the consumer recovers, nacks and continues or lets the process die. Whichever you choose, write the test, because the default of crashing and restarting in a loop is what happens when nobody decides.
• Poison messages. A message that always fails transiently will be requeued forever. Count attempts in the delivery, add an Attempt field, and test that the consumer gives up after a limit by settling with Nack(false). The retry testing article shows how to do that without waiting.
• Concurrency. If the consumer runs several handlers at once, test that settling is safe from more than one goroutine. The recorder in this lab is not locked and would be caught by the race detector, so add a mutex to it when you add workers.
• Ordering. If the broker guarantees per partition ordering and your code relies on it, the test should deliver messages for two keys interleaved and assert that each key's messages stay in order.

Where the fake stops

The fake proves your decisions. It says nothing about the adapter, which is the code that turns broker messages into Delivery values and calls the real Ack. Mistakes live there: acknowledging before the handler finishes, using the wrong requeue flag, losing a message when the channel is full, or never closing the channel on shutdown. Test the adapter with the real broker in a container, using testcontainers, with a handful of cases: publish, consume, nack with requeue, and a restart. Keep that suite small, since each case takes seconds.

Delivery guarantees differ by broker, and the test should encode the one you use. The RabbitMQ acknowledgement documentation and the Go blog on pipelines and cancellation are good references for the two halves of this design: what the broker promises, and how to stop a channel based loop cleanly.