Stage 7 · Master
Phase 16 — Kafka Integration
Retry
Distinguish a malformed event from a database that is momentarily down, and give the consumer a bounded, in-loop retry that recovers from the second without ever retrying the first.
handler.HandlePaymentCompleted's Single error Return Hides Two Unrelated Failures
Lesson 3's HandlePaymentCompleted returns a plain error in two situations that demand opposite responses. If json.Unmarshal fails, the message itself is malformed — retrying it a thousand times produces the same failure a thousand times; the only sane action is to stop retrying and record it. If tx.Exec fails because notification-service's database is briefly unreachable, the event is perfectly valid — retrying after a short wait is exactly the right action. Run currently retries neither case; it just returns the error and the process exits. This lesson gives the loop a way to tell the two apart.
ErrPoisonEvent Marks the First Category Explicitly
package consumer
import (
"errors"
"fmt"
)
// ErrPoisonEvent marks a message that will never succeed no matter how many
// times it is retried — a malformed payload or an unsupported event_version
// (Lesson 7 adds the second check). Wrapping a cause with errPoisonEvent lets
// the retry loop use errors.Is to skip straight past it, while %w preserves
// the underlying cause for logs.
var ErrPoisonEvent = errors.New("consumer: poison event")
func errPoisonEvent(cause error) error {
return fmt.Errorf("%w: %v", ErrPoisonEvent, cause)
}
The verbs are not interchangeable here. Wrapping the cause with a second %w would let errors.Is(err, someTransientError) match a message that is definitively poison, and the retry loop would keep re-delivering it. Only the sentinel is wrapped; the cause is rendered for the log line and nothing more.
Lesson 3's unmarshal-error line changes from returning the bare error to wrapping it, marking it poison rather than transient:
var env paymentCompletedEnvelope
if err := json.Unmarshal(raw, &env); err != nil {
return errPoisonEvent(err)
}
processWithRetry Wraps the Handler, Not the Whole Loop
package consumer
import (
"context"
"errors"
"time"
"github.com/rs/zerolog/log"
)
const maxAttempts = 5
// processWithRetry retries transient failures with capped exponential
// backoff, doubling from 1s and never waiting longer than 30s between
// attempts. A poison event short-circuits immediately on the first
// attempt — retrying it would only delay the DLQ path Lesson 6 adds.
func processWithRetry(ctx context.Context, handler *Handler, topic string, value []byte) error {
var lastErr error
wait := time.Second
for attempt := 1; attempt <= maxAttempts; attempt++ {
err := handler.HandlePaymentCompleted(ctx, topic, value)
if err == nil {
return nil
}
if errors.Is(err, ErrPoisonEvent) {
return err
}
lastErr = err
log.Warn().
Int("attempt", attempt).
Int("max_attempts", maxAttempts).
Dur("wait", wait).
Err(err).
Msg("consumer: transient handler error, retrying")
if attempt == maxAttempts {
break
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(wait):
}
wait *= 2
if wait > 30*time.Second {
wait = 30 * time.Second
}
}
return lastErr
}
if err := processWithRetry(ctx, handler, msg.Topic, msg.Value); err != nil {
return err // Lesson 6 routes this to the DLQ instead of exiting the process
}
processWithRetry runs before CommitMessages and before the next FetchMessage. While it waits out its backoff, every other message queued behind it on that same partition sits unprocessed too — a burst of transient failures on one partition can visibly increase end-to-end latency for that partition's other, perfectly healthy events. This is the direct cost of manual, ordered commit from Lesson 4; it is accepted deliberately in exchange for never silently dropping a message.
Simulate a Down Database and Watch the Backoff, Then Recovery
Applied exercise
Prove a poison event never enters the retry loop at all
processWithRetry is supposed to short-circuit poison events on attempt 1 rather than waiting through all 5 attempts. Prove it does.
- Publish a message to payment.completed whose payload is not valid JSON, e.g. a bare string 'not-json'.
- Time how long it takes the consumer's log line for that message to appear, from publish to log.
- Confirm it appears almost instantly, not after ~1s + 2s + 4s + 8s of backoff.
- Confirm the log line's error mentions ErrPoisonEvent (or its wrapped message), not a database connection error.
Deliverable
The observed log timestamps proving the short-circuit, plus the log line's exact text.
Completion checks
- The poison event is logged within roughly the time it takes to fetch and unmarshal it — no backoff delay observed.
- The log clearly attributes the failure to a poison/malformed event, not a transient one.
Why does processWithRetry check errors.Is(err, ErrPoisonEvent) and return immediately, instead of letting a malformed event go through all 5 attempts like any other error?
Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.