shunt

README

shunt

A Kafka router for Go, built on IBM/sarama: register a handler per topic and get consumer-group consumption, composable retry chains (in-process and retry-topic based), dead-letter queues, and bounded worker pools — with at-least-once delivery and per-partition ordering on the happy path.

Features

• Per-topic handlers — one router, one consumer group, a handler per topic.
• Retry chains — each topic composes its own escalation policy from two step types:
  • InProcess(n) / InProcessWithBackoff(n, cfg) — inline retries with exponential backoff and jitter.
  • RetryTopic(name, delay, attempts) — republish to a dedicated retry topic and re-attempt after a delay; the router consumes retry topics itself.
• DLQ — after the whole chain is exhausted, the message lands in a per-topic dead-letter topic with full failure metadata (or is dropped loudly if no DLQ is configured).
• Worker pools — WithWorkers(n) bounds handler concurrency router-wide; WithTopicWorkers(n) additionally caps one topic (including its retry topics) so a heavy topic can't starve the rest.
• At-least-once — offsets are committed only after a message is fully dealt with; messages that succeed on the main topic are processed in order; anything sent to a retry topic is reordered by design.

Install

go get github.com/devilreza/shunt

Quickstart

router, err := shunt.New([]string{"localhost:9092"}, "billing",
	shunt.WithWorkers(8),
)
if err != nil {
	log.Fatal(err)
}

router.Register(
	shunt.NewTopic("orders", handleOrder,
		shunt.WithRetry(
			shunt.InProcess(2),                                  // 2 inline retries with backoff
			shunt.RetryTopic("orders.retry.10s", 10*time.Second, 2), // then 2 delayed attempts
			shunt.RetryTopic("orders.retry.5m", 5*time.Minute, 1),   // then 1 last, much later
		),
		shunt.WithDLQ("orders.dlq"),
		shunt.WithTopicWorkers(2),
	),
)

go func() { _ = router.Run() }()
// ...
_ = router.Shutdown(ctx)

A handler is just:

func handleOrder(ctx context.Context, msg *shunt.Message) error {
	// msg.Retry is non-nil on retry deliveries (attempt counts, original coordinates, last error).
	return process(msg.Value)
}

See examples/consumer for a runnable demo with graceful shutdown.

How a message moves through the chain

For the orders topic above, a message that keeps failing travels:

orders ──1 + 2 inline attempts──▶ orders.retry.10s ──wait 10s, attempt──▶ (×2)
       └─────────────────────────▶ orders.retry.5m  ──wait 5m,  attempt──▶ orders.dlq

1. The initial delivery runs the handler once; InProcess(2) adds up to 2 inline attempts with backoff.
2. Still failing, the message is republished to orders.retry.10s. The router consumes that topic too, waits until the message's not-before instant, and re-runs the handler — up to 2 deliveries for this step.
3. Same for orders.retry.5m with a longer delay.
4. Chain exhausted: the message is published to orders.dlq (or dropped with an ERROR log if no DLQ is configured).

Register adds every retry topic in a chain to the routing table automatically — you register orders once and the router subscribes to orders, orders.retry.10s, and orders.retry.5m in the same consumer group.

Waiting messages don't hold worker slots, and a delay only blocks its own (dedicated) retry-topic partition. InProcess steps may also appear mid-chain, between retry topics.

Retry headers

Retry state travels in record headers; user headers are preserved verbatim and stale x-shunt-* headers are re-stamped on every hop. The original key is kept, so key affinity survives across retry topics.

Header	Meaning
x-shunt-original-topic / -partition / -offset	Coordinates of the first delivery on the main topic
x-shunt-attempts	Total handler invocations so far
x-shunt-step	Index of the chain step that owns the delivering retry topic
x-shunt-step-attempt	1-based delivery number within that step
x-shunt-not-before	RFC3339Nano instant before which the handler must not run
x-shunt-last-error	Last handler error (truncated to 1 KiB)
x-shunt-error, x-shunt-failed-at	DLQ records only: final error and failure time

Handlers see this parsed on Message.Retry. Foreign or corrupt headers on a retry topic are tolerated: the route determines the chain position, and the message is treated as ready to run.

Delivery semantics

Delivery is at-least-once. An offset is marked only after one of:

• the handler succeeds,
• the message is durably republished to the next retry hop (RequiredAcks = WaitForAll),
• the message is durably published to the DLQ,
• the chain is exhausted with no DLQ (logged at ERROR and dropped, so a poison message cannot stall its partition).

If a republish fails, the offset is not marked and Kafka redelivers — which can repeat a delivery's inline attempts. A rebalance or crash mid-wait redelivers parked retry messages; the wait is computed from the absolute not-before header, so they simply wait out the remainder. Handlers must be idempotent.

Ordering: messages that succeed on the main topic are processed in order; anything sent to a retry topic is reordered by design. A hop moves the message to a different topic and partition and delays it, so later messages from the main partition can (and will) overtake it. If strict per-key ordering matters even under failure, use only InProcess steps — inline retries block the partition and preserve order.

Two caveats worth knowing:

• A retry-topic delay holds its offset uncommitted for the full wait; after a rebalance, every parked message in the claim is redelivered and re-waits.
• Waits are capped at the step's configured delay, so a skewed or bogus not-before cannot park a partition beyond its policy.

Observability (OpenTelemetry)

Tracing and metrics are opt-in and independent; with neither enabled shunt adds zero overhead and touches no headers. Only the lightweight OTel API is a library dependency — you bring the SDK.

router, _ := shunt.New(brokers, "billing",
	shunt.WithTracing(tracerProvider), // or shunt.WithTracing() for the OTel global
	shunt.WithMetrics(meterProvider),  // or shunt.WithMetrics() for the OTel global
)

Tracing creates one consumer span per delivery (process <topic>, handed to your handler via ctx) and one producer span per retry-hop or DLQ publish (send <topic>). Trace context is propagated through traceparent/tracestate/baggage record headers, so a message's whole journey is a single connected trace:

process orders ─▶ send orders.retry.10s ┈(delay)┈▶ process orders.retry.10s ─▶ send orders.dlq

Failed inline attempts appear as exception events on the process span; spans carry Kafka messaging attributes (partition, offset, consumer group) plus shunt.step, shunt.attempts, and shunt.outcome (success|retried|dlq|dropped|redeliver). The default propagator is W3C TraceContext+Baggage — deliberately not the OTel global, which is a no-op unless configured; use WithPropagator to override.

Metrics registers these instruments (scope github.com/devilreza/shunt), labeled only with bounded attributes — topic, outcome, publish kind, error flag; never partition or offset:

Metric	Type	Description
shunt.messages.processed	counter	Deliveries completed, by topic and outcome
shunt.handler.duration	histogram (s)	Duration of individual handler attempts
shunt.messages.published	counter	Retry-hop and DLQ publishes
shunt.messages.inflight	up-down counter	Deliveries currently holding a worker slot

See examples/otel for a runnable setup with stdout exporters.

Topics are not auto-created

shunt never creates topics. Before Run, make sure the main topic, every RetryTopic name in the chain, and the DLQ topic exist (or your cluster allows auto-creation). Retry topics are ordinary topics; one partition is often enough since throughput is low and delays dominate.

Configuration

shunt.New accepts:

Option	Effect
WithWorkers(n)	Router-wide handler concurrency bound (default 10)
WithLogger(l)	*slog.Logger to use (default slog.Default())
WithSaramaConfig(cfg)	Full sarama control (TLS, SASL, versions, offsets). shunt still forces consumer error reporting and, when anything publishes, durable synchronous producing

shunt.NewTopic accepts WithRetry(steps...), WithDLQ(topic), and WithTopicWorkers(n). The default retry policy is InProcess(3) with a 1s → 30s backoff. WithRetry() with no steps disables retries: one failure goes straight to the DLQ.

Development

make build   # go build ./...
make test    # go test -race ./...
make lint    # golangci-lint run

Commits follow Conventional Commits (see .cz.toml).

License

MIT — see LICENSE.

Documentation

Overview

Package shunt routes Kafka messages from a consumer group to per-topic handlers, with a bounded worker pool, per-topic retry chains, and an optional dead-letter topic.

A topic's retry policy is an ordered chain of steps. InProcess steps retry the handler inline with exponential backoff; RetryTopic steps republish the failed message to a dedicated retry topic and re-attempt it after a configurable delay. Steps compose freely, so a topic can escalate from cheap inline retries to progressively longer-delayed retry topics before giving up:

shunt.NewTopic("orders", handleOrder,
	shunt.WithRetry(
		shunt.InProcess(2),
		shunt.RetryTopic("orders.retry.10s", 10*time.Second, 2),
		shunt.RetryTopic("orders.retry.5m", 5*time.Minute, 1),
	),
	shunt.WithDLQ("orders.dlq"),
)

Retry topics are consumed by the same router and consumer group as the main topic; Register adds them to the routing table automatically. Retry state (original coordinates, attempt counts, the not-before instant) travels in x-shunt-* record headers, and handlers can inspect it via Message.Retry. shunt does not create topics: main, retry, and DLQ topics must exist before Run (or the cluster must allow auto-creation).

Delivery is at-least-once: an offset is marked only after the handler succeeds, the message is republished to the next retry hop, it lands in the DLQ, or the chain is exhausted with no DLQ configured (logged and dropped so a poison message cannot stall its partition). A crash or rebalance can redeliver messages that were already processed but not yet committed — and can repeat a delivery's inline attempts — so handlers must be idempotent.

Messages within one partition are processed strictly in order; parallelism comes from distinct partitions, capped globally by WithWorkers and optionally per topic by WithTopicWorkers. Messages waiting out a retry delay do not hold worker slots. Ordering holds only for messages that succeed on the main topic: anything sent to a retry topic is reordered by design, since a hop moves the message to another topic and delays it while later messages from the main partition keep flowing. Chains built solely from InProcess steps retry inline and preserve ordering under failure.

Observability is opt-in via OpenTelemetry and each signal is independent. WithTracing opens one consumer span per delivery ("process <topic>", visible to the handler through its context) and one producer span per retry-hop or DLQ publish ("send <topic>"); trace context travels across hops in W3C traceparent/tracestate/baggage record headers, so a message's whole retry chain forms a single connected trace, and failed attempts are recorded as exception events. WithMetrics registers the default instruments under the "github.com/devilreza/shunt" scope:

• shunt.messages.processed — counter of completed deliveries, by registered topic and outcome (success, retried, dlq, dropped, redeliver, canceled)
• shunt.handler.duration — histogram (seconds) of individual handler attempts, by topic and error flag
• shunt.messages.published — counter of retry-hop and DLQ publishes, by topic, kind, and error flag
• shunt.messages.inflight — up-down counter of deliveries holding a worker slot, by topic

Metric attributes are restricted to bounded sets; partition, offset, and error details appear only on spans. Both options accept a provider or default to the OTel globals, and WithPropagator overrides header propagation. With neither option set, shunt adds no overhead.

Usage:

router, err := shunt.New(brokers, "billing", shunt.WithWorkers(8))
if err != nil { ... }
router.Register(
	shunt.NewTopic("invoice.created", handleInvoice,
		shunt.WithRetry(shunt.InProcess(3), shunt.RetryTopic("invoice.created.retry", time.Minute, 3)),
		shunt.WithDLQ("invoice.created.dlq"),
		shunt.WithTopicWorkers(2),
	),
)
go func() { _ = router.Run() }()
...
_ = router.Shutdown(ctx)

Index

• Variables
• type BackoffConfig
• type Claim
• type HandlerFunc
• type Header
• type Message
• type Producer
• type RetryMeta
• type RetryStep
  • func InProcess(attempts int) RetryStep
  • func InProcessWithBackoff(attempts int, cfg BackoffConfig) RetryStep
  • func RetryTopic(name string, delay time.Duration, attempts int) RetryStep
• type Router
  • func New(brokers []string, groupID string, opts ...RouterOption) (*Router, error)
  • func (r *Router) Register(topics ...*Topic) *Router
  • func (r *Router) Run() error
  • func (r *Router) Shutdown(ctx context.Context) error
• type RouterOption
  • func WithLogger(logger *slog.Logger) RouterOption
  • func WithMetrics(mp ...metric.MeterProvider) RouterOption
  • func WithPropagator(p propagation.TextMapPropagator) RouterOption
  • func WithSaramaConfig(cfg *sarama.Config) RouterOption
  • func WithTracing(tp ...trace.TracerProvider) RouterOption
  • func WithWorkers(n int) RouterOption
• type Session
• type Topic
  • func NewTopic(name string, handler HandlerFunc, opts ...TopicOption) *Topic
  • func (t *Topic) Name() string
• type TopicOption
  • func WithDLQ(topic string) TopicOption
  • func WithRetry(steps ...RetryStep) TopicOption
  • func WithTopicWorkers(n int) TopicOption

Variables

var (
	// ErrAlreadyRunning is returned by Run when the router is already consuming.
	ErrAlreadyRunning = errors.New("shunt: router is already running")
	// ErrNoTopics is returned by Run when no topics have been registered.
	ErrNoTopics = errors.New("shunt: no topics registered")
)

Types

type BackoffConfig

type BackoffConfig struct {
	Initial    time.Duration
	Max        time.Duration
	Multiplier float64
	Jitter     float64
}

BackoffConfig shapes the exponential backoff between in-process retry attempts.

type Claim

type Claim interface {
	Topic() string
	Partition() int32
	Messages() <-chan *sarama.ConsumerMessage
}

Claim is the narrow slice of sarama.ConsumerGroupClaim the router needs.

type HandlerFunc

type HandlerFunc func(ctx context.Context, msg *Message) error

HandlerFunc processes one message; returning an error triggers the topic's retry chain and DLQ policy.

type Header

type Header struct {
	Key   string
	Value []byte
}

Header is a single Kafka record header.

type Message

type Message struct {
	Topic     string
	Partition int32
	Offset    int64
	Key       []byte
	Value     []byte
	Timestamp time.Time
	Headers   []Header

	// Retry is non-nil when this delivery arrived through a retry topic; it
	// describes where the message sits in its topic's retry chain.
	Retry *RetryMeta
}

Message is a Kafka record decoupled from sarama, as delivered to handlers.

type Producer

type Producer interface {
	SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error)
	Close() error
}

Producer is the seam over sarama.SyncProducer used for retry-topic and DLQ publishing; sarama.SyncProducer satisfies it structurally.

type RetryMeta

type RetryMeta struct {
	// OriginalTopic, OriginalPartition, and OriginalOffset locate the very
	// first delivery of this message on its main topic.
	OriginalTopic     string
	OriginalPartition int32
	OriginalOffset    int64
	// Attempts is how many times a handler was invoked before this delivery.
	Attempts int
	// Step is the index of the RetryTopic step in the chain that delivered
	// this message.
	Step int
	// StepAttempt is the 1-based delivery number within that step.
	StepAttempt int
	// NotBefore is the instant before which the handler must not run.
	NotBefore time.Time
	// LastError is the (possibly truncated) error from the previous attempt.
	LastError string
}

RetryMeta is the retry state carried in x-shunt-* headers across hops.

type RetryStep

type RetryStep struct {
	// contains filtered or unexported fields
}

RetryStep is one stage of a topic's retry chain; build steps with InProcess, InProcessWithBackoff, or RetryTopic and compose them in order with WithRetry.

func InProcess

func InProcess(attempts int) RetryStep

InProcess retries the handler up to attempts more times inline, with the default exponential backoff (1s initial, 30s cap, x2, 0.2 jitter).

func InProcessWithBackoff

func InProcessWithBackoff(attempts int, cfg BackoffConfig) RetryStep

InProcessWithBackoff retries the handler up to attempts more times inline, waiting per cfg between attempts. Inline waits block the message's partition, so keep cfg.Max modest and use RetryTopic steps for long delays.

func RetryTopic

func RetryTopic(name string, delay time.Duration, attempts int) RetryStep

RetryTopic republishes a failed message to the named topic and re-attempts it after delay, up to attempts deliveries. The router consumes the topic itself; it must not be shared with any other registered topic or chain.

type Router

type Router struct {
	// contains filtered or unexported fields
}

Router consumes from a Kafka consumer group and dispatches each message to the Topic registered for it, bounding concurrency with worker pools and driving each topic's retry chain and DLQ.

func New

func New(brokers []string, groupID string, opts ...RouterOption) (*Router, error)

New builds a Router; it validates its arguments and performs no I/O.

func (*Router) Register

func (r *Router) Register(topics ...*Topic) *Router

Register adds topics — and every retry topic in their chains — to the routing table. Like gin route registration, misuse is a programmer error at boot time and panics: nil topic, empty name or handler, an invalid retry chain, any name collision, or registering after Run.

func (*Router) Run

func (r *Router) Run() error

Run connects to the brokers and consumes until Shutdown is called. It blocks and returns nil on clean shutdown.

func (*Router) Shutdown

func (r *Router) Shutdown(ctx context.Context) error

Shutdown stops consuming, waits for in-flight handlers to finish (they see a canceled context), and closes the group and producer. Calling it before Run, or twice, is a no-op.

type RouterOption

type RouterOption func(*Router)

RouterOption customizes a Router at construction time.

func WithLogger

func WithLogger(logger *slog.Logger) RouterOption

WithLogger sets the router's logger; slog.Default() is used otherwise. A nil logger is ignored.

func WithMetrics

func WithMetrics(mp ...metric.MeterProvider) RouterOption

WithMetrics enables the router's default OpenTelemetry metrics — shunt.messages.processed, shunt.handler.duration, shunt.messages.published, and shunt.messages.inflight — labeled only with bounded attributes (registered topic, outcome, publish kind, error flag). With no argument the global otel.GetMeterProvider() is used, resolved at Run.

func WithPropagator

func WithPropagator(p propagation.TextMapPropagator) RouterOption

WithPropagator overrides the propagator used to extract and inject trace context on record headers. The default is a W3C TraceContext+Baggage composite — deliberately not otel.GetTextMapPropagator(), whose global default is a no-op that would silently disconnect traces across retry hops; pass otel.GetTextMapPropagator() explicitly to use the global.

func WithSaramaConfig

func WithSaramaConfig(cfg *sarama.Config) RouterOption

WithSaramaConfig replaces the router's default sarama config, for brokers that need TLS/SASL, a specific version, or tuned offsets. The router still forces the few fields it depends on: consumer error reporting and, when any topic publishes to a retry topic or DLQ, durable synchronous producing.

func WithTracing

func WithTracing(tp ...trace.TracerProvider) RouterOption

WithTracing enables OpenTelemetry tracing: one consumer span per delivery (the handler receives its context), a producer span per retry-hop or DLQ publish, and W3C trace-context propagation through record headers so a message's whole retry chain forms one connected trace. With no argument the global otel.GetTracerProvider() is used, resolved at Run.

func WithWorkers

func WithWorkers(n int) RouterOption

WithWorkers bounds how many handlers run concurrently across all claimed partitions; messages within one partition are always processed in order.

type Session

type Session interface {
	Context() context.Context
	MarkMessage(msg *sarama.ConsumerMessage, metadata string)
}

Session is the narrow slice of sarama.ConsumerGroupSession the router needs; sarama's interface satisfies it structurally.

type Topic

type Topic struct {
	// contains filtered or unexported fields
}

Topic couples a Kafka topic name with its handler, retry chain, DLQ, and worker policy.

func NewTopic

func NewTopic(name string, handler HandlerFunc, opts ...TopicOption) *Topic

NewTopic builds a Topic with the default retry chain (InProcess(3)), then applies the given options.

func (*Topic) Name

func (t *Topic) Name() string

Name returns the Kafka topic name.

type TopicOption

type TopicOption func(*Topic)

TopicOption customizes a Topic at construction time.

func WithDLQ

func WithDLQ(topic string) TopicOption

WithDLQ publishes a message to the given dead-letter topic once the whole retry chain is exhausted; without it the message is dropped after the last attempt.

func WithRetry

func WithRetry(steps ...RetryStep) TopicOption

WithRetry replaces the topic's default retry chain with the given steps, executed in order after the initial delivery fails. WithRetry() with no steps disables retries entirely: a single failure goes straight to the DLQ (or is dropped).

func WithTopicWorkers

func WithTopicWorkers(n int) TopicOption

WithTopicWorkers caps how many handlers run concurrently for this topic, counting its retry topics against the same cap, so one heavy topic cannot starve the router's shared pool.