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.
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 within a partition are processed strictly in order.
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
- The initial delivery runs the handler once;
InProcess(2)adds up to 2 inline attempts with backoff. - Still failing, the message is republished to
orders.retry.10s. The router consumes that topic too, waits until the message'snot-beforeinstant, and re-runs the handler — up to 2 deliveries for this step. - Same for
orders.retry.5mwith a longer delay. - 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.
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-beforecannot park a partition beyond its policy.
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.
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 ¶
Constants ¶
This section is empty.
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") )
Functions ¶
This section is empty.
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 ¶
HandlerFunc processes one message; returning an error triggers the topic's retry chain and DLQ policy.
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 ¶
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 ¶
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 ¶
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.
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 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 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.
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.
Source Files
Directories