Delivery Guarantees

At-most-once, at-least-once, why exactly-once is not available end to end, and how to make duplicates harmless.

Syntaxack after the work -> at-least-once; ack before the work -> at-most-once

A delivery guarantee is a promise about how many times a message reaches your code. There are three names in use. Two of them describe something you can actually have.

At-most-once

The consumer tells the broker it is finished with the message before doing the work. Telling the broker is called acknowledging, usually shortened to acking; in Kafka it is committing the offset. If the process dies next, nobody retries and the message is gone. Choose this only when losing one costs less than handling one twice, such as a metrics sample.

At-least-once

The consumer does the work first and acks after. If it dies in between, the broker never heard the ack and delivers the message again. Nothing is lost, and duplicates happen. This is the default in Kafka, NATS JetStream, RabbitMQ and SQS, and it is the one you want.

Exactly-once

No broker can promise this end to end. The ack and the side effect are two separate actions, and a crash can land between them whichever order you put them in. Kafka transactions get close inside Kafka: a transactional producer writes messages and consumer offsets as one unit, and a consumer set to isolation.level=read_committed never sees an aborted batch. That covers Kafka-to-Kafka work only. It cannot unsend an email.

What to build instead

Accept at-least-once and make a second delivery harmless. Give every message an id, keep a record of ids you have applied, and write that record in the same database transaction as the work itself. A duplicate then changes nothing. That property is called idempotency, and it is the real answer to exactly-once.

Example

Example · go
package main

import "fmt"

// processor applies each message id once, however often it arrives.
type processor struct {
	applied map[string]bool // ids already handled
	balance int             // the state we are protecting
}

func (p *processor) handle(id string, cents int) {
	if p.applied[id] {
		// A redelivery: the work was already done, so do nothing.
		fmt.Println("duplicate ignored:", id)
		return
	}
	p.applied[id] = true
	p.balance += cents
	fmt.Println("applied:", id)
}

func main() {
	p := &processor{applied: map[string]bool{}}
	// The broker delivers msg-1 twice because the first ack was lost.
	p.handle("msg-1", 10)
	p.handle("msg-1", 10)
	p.handle("msg-2", 5)
	fmt.Println("balance:", p.balance)
}

// Output:
// applied: msg-1
// duplicate ignored: msg-1
// applied: msg-2
// balance: 15

When to use it

  • Decide where to commit the Kafka offset in a worker that charges customers.
  • Add a processed_messages table so a redelivered payment event cannot double-credit an account.
  • Explain to a reviewer why the service does not claim exactly-once delivery.
  • Stop one poisonous message from blocking a partition by moving it to a dead-letter topic.

More examples

Where the commit goes decides the guarantee

The guarantee is not a setting, it is the position of one line. Keep the commit after the work in every consumer you write, and treat the at-most-once variant as something you do on purpose for cheap, high-volume data. Note that the failure path does not commit either, which is exactly how a retry happens.

Example · go
package main

import (
	"context"
	"log"

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

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

	ctx := context.Background()
	for {
		m, err := r.FetchMessage(ctx) // reads without committing
		if err != nil {
			log.Fatal(err)
		}

		// At-most-once would commit here, before the work: a crash on the
		// next line loses this message for good.
		if err := sendEmail(m.Value); err != nil {
			// No commit, so the broker hands this message over again.
			log.Print(err)
			continue
		}
		// At-least-once commits after the work: a crash before this line
		// means the message arrives a second time.
		if err := r.CommitMessages(ctx, m); err != nil {
			log.Fatal(err)
		}
	}
}

func sendEmail(body []byte) error {
	log.Printf("sending %s", body)
	return nil
}

A duplicate check in the same transaction as the work

The marker row and the balance change are in one transaction, so there is no window where the message counts as handled but the money has not moved. A separate seen-set in memory, or in Redis, does not give you that. Use this in any consumer that touches money, stock levels or anything else a customer would notice twice.

Example · go
package main

import (
	"context"
	"database/sql"
)

// Schema:
//   CREATE TABLE processed_messages (id text PRIMARY KEY);

// creditAccount applies one payment message, even if it is delivered twice.
func creditAccount(ctx context.Context, db *sql.DB, msgID string, cents int64) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback() // does nothing once the transaction is committed

	// The primary key turns a second insert of the same id into zero rows.
	res, err := tx.ExecContext(ctx,
		`INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING`, msgID)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return nil // an earlier delivery already applied this message
	}

	if _, err := tx.ExecContext(ctx,
		`UPDATE accounts SET balance = balance + $1 WHERE id = $2`, cents, 1); err != nil {
		return err
	}
	return tx.Commit() // the marker and the money commit together or not at all
}

Give up after a fixed number of attempts

At-least-once without a limit is an infinite loop: one message that can never succeed blocks every message behind it in its partition. A dead-letter topic is a second topic that holds the failures so a human can look at them. Keep the attempt count in the message headers or your own table, because the broker does not count for you.

Example · go
package main

import (
	"context"
	"errors"
	"log"

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

// apply is the real handler; it is the part that can fail.
func apply(ctx context.Context, m kafka.Message) error {
	return errors.New("downstream unavailable")
}

// handle retries a message a few times, then parks it on a dead-letter topic.
func handle(ctx context.Context, dlq *kafka.Writer, m kafka.Message, attempt int) error {
	err := apply(ctx, m)
	if err == nil {
		return nil
	}
	if attempt < 5 {
		return err // do not commit: the message comes back
	}
	// Five failures is not a passing glitch. Park the message, then commit,
	// otherwise this partition never moves past it.
	log.Printf("parking %s after %d attempts: %v", m.Key, attempt, err)
	return dlq.WriteMessages(ctx, kafka.Message{Key: m.Key, Value: m.Value})
}

Related

go job idempotencygo outboxgo retriesgo messaging basicsgo sql transactions

Discussion

  • Be the first to comment on this lesson.