README
¶
BabelQueue Kafka transport (Go)
A Kafka-backed babelqueue.Transport for the BabelQueue Go runtime,
built on segmentio/kafka-go (pure Go, no cgo).
go get github.com/babelqueue/babelqueue-go/kafka
Retry tiers and the re-inject consumer (ADR-0034)
Kafka has no delayed delivery. The retry-topic pattern parks a record on a tier topic
<topic>.retry.<n> (n starts at 1) and a consumer carries it back once the tier has elapsed.
tr, _ := kafka.New(
kafka.WithBrokers("localhost:9092"),
kafka.WithGroupID("orders-reinject"), // use a dedicated group for the re-injector
kafka.WithRetryTiers(kafka.DefaultRetryTiers...), // 5s, 30s, 5m, 30m (RECOMMENDED, opt-in)
)
reinjector, _ := tr.NewReinjector("orders")
go reinjector.Run(ctx) // without it a retried or delayed record is never redelivered
- Fail target:
PublishRetryForAttempt(ctx, from, attemptsBefore)writes to<topic>.retry.<min(attemptsBefore, n-1)+1>; once the tiers run out the last tier is kept. It setsbq-attempts=attemptsBefore+1, syncs the bodyattempts, setsbq-delay(the tier in ms) andbq-original-topic, and keeps the record key byte for byte. - Delay request:
PublishDelayed/PublishRetry(ctx, from, delay)pick the smallest tier that is=
delay, and returnErrDelayUnsupportedwhen none fits (no clamping).PublishRetryis deprecated in favour ofPublishRetryForAttempt; its behaviour is unchanged. retry_backoffdoes not apply on Kafka: the tier sets the delay.- Re-inject consumer: for each record of
<topic>.retry.<n>it waits untilrecord.timestamp + tier_ms(clamped to[now, now+tier_ms], so an old record is carried at once), writes the value and key byte for byte tobq-original-topic(no decode, no re-encode,attemptsuntouched, an undecodable record is carried as it is), dropsbq-delayandbq-original-topic, keeps every other header (traceparentincluded) and commits the retry-topic offset only after the broker acknowledged the write (at-least-once). kafka-go has no partition pause/resume for a group reader, so the wait is held per partition in a worker fed through a bounded buffer: the fetch loop keeps polling while a record sleeps. - Create the retry topics with
message.timestamp.type=LogAppendTime(SHOULD). The wait is anchored on the record timestamp; withLogAppendTimeit is the broker's append time, so a producer's clock cannot skew the delay. With the defaultCreateTimethe producer's clock is trusted. - Poison records (identity-less or undecodable envelopes) go to
<topic>.dlq; the identity rule follows the conformancekafka_poisoncases.
Full spec: babelqueue.com · MIT © Muhammet Şafak
Documentation
¶
Overview ¶
Package kafka is an Apache Kafka-backed babelqueue.Transport for the BabelQueue Go runtime, on the pure-Go github.com/segmentio/kafka-go (no CGo, no librdkafka). Producing writes the canonical envelope as the record value and projects the contract fields onto native Kafka record headers (bq-job = URN, bq-trace-id, bq-message-id, bq-schema-version, bq-source-lang, bq-attempts — all UTF-8 byte strings) with the record timestamp mirroring meta.created_at, so a Java/.NET/... peer routes on bq-job without parsing the body. Consuming is process-then-commit: FetchMessage reserves a record (no auto-commit), and Ack commits the offset only after the handler returns (at-least-once). Kafka has no native delivery count, so the bq-attempts header is the authoritative retry counter (the body's attempts is the fallback for non-BabelQueue producers); the App owns retry by republishing with attempts+1 and dead-letters to <queue>.dlq.
tr, _ := kafka.New(kafka.WithBrokers("localhost:9092"), kafka.WithGroupID("orders-workers"))
app := babelqueue.NewApp(tr, babelqueue.WithDefaultQueue("orders"))
A record whose bq-schema-version header is present, not blank and not the core's supported version is version-gated (§6.3 / §6.7): the body is not decoded and the record is routed to a reserved unsupported-version URN no handler owns, so the App dead-letters it directly, with no retry (whatever the unknown-URN strategy, StrategyRelease included), or drops it when dead-lettering is off. The optional OnSchemaVersionRejected hook names the rejected version.
This binding implements §6 of the BabelQueue broker-bindings contract. The envelope is unchanged (schema_version stays 1); Apache Kafka is purely additive.
Full spec: https://babelqueue.com
Index ¶
- Variables
- func RetryTopic(topic string, n int) string
- type OnSchemaVersionRejected
- type Option
- func WithBrokers(brokers ...string) Option
- func WithGroupID(groupID string) Option
- func WithMaxWaitTime(d time.Duration) Option
- func WithOnSchemaVersionRejected(fn OnSchemaVersionRejected) Option
- func WithReaderFactory(f func(queue string) Reader) Option
- func WithRetryTiers(tiers ...time.Duration) Option
- func WithWriter(w Writer) Option
- type Reader
- type Reinjector
- type Transport
- func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
- func (t *Transport) Close() error
- func (t *Transport) NewReinjector(workTopics ...string) (*Reinjector, error)
- func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
- func (t *Transport) Publish(ctx context.Context, queue, body string) error
- func (t *Transport) PublishDelayed(ctx context.Context, queue, body string, delay time.Duration) error
- func (t *Transport) PublishFrom(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage) error
- func (t *Transport) PublishFromWithHeaders(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage, ...) error
- func (t *Transport) PublishRetry(ctx context.Context, from *babelqueue.ReceivedMessage, delay time.Duration) errordeprecated
- func (t *Transport) PublishRetryForAttempt(ctx context.Context, from *babelqueue.ReceivedMessage, attemptsBefore int) error
- func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
- func (t *Transport) Rewind(_ context.Context, msg *babelqueue.ReceivedMessage) error
- type Writer
Constants ¶
This section is empty.
Variables ¶
var DefaultRetryTiers = []time.Duration{5 * time.Second, 30 * time.Second, 5 * time.Minute, 30 * time.Minute}
DefaultRetryTiers is the RECOMMENDED retry tier list of broker-bindings.md §6.4 (5s, 30s, 5m, 30m), the Go counterpart of the contract's DEFAULT_RETRY_TIERS_MS constant. It is opt-in: no tiers are configured unless the caller passes it (or their own list) to WithRetryTiers. Use DefaultRetryTiersMs for the millisecond values.
var DefaultRetryTiersMs = []int64{5000, 30000, 300000, 1800000}
DefaultRetryTiersMs is DefaultRetryTiers in milliseconds: [5000, 30000, 300000, 1800000].
var ErrClosed = errors.New("kafka: transport is closed")
ErrClosed is returned by Pop and Ack once Transport.Close has run: a closed transport never opens a new consumer-group reader (it would rejoin the group and never be closed).
var ErrDelayUnsupported = errors.New("kafka: delay needs a retry tier that covers it (§6.4)")
ErrDelayUnsupported is returned by Transport.PublishDelayed and Transport.PublishRetry when the request cannot be honoured: no retry tiers are configured (WithRetryTiers), or the requested delay exceeds the largest tier. Kafka has no native delayed delivery, so a delay is never silently dropped or clamped (broker-bindings.md §6.4).
Functions ¶
func RetryTopic ¶ added in v1.4.0
RetryTopic returns the retry topic name of the 1-based tier n for work topic topic.
Types ¶
type OnSchemaVersionRejected ¶ added in v1.1.0
type OnSchemaVersionRejected func(version, queue string)
OnSchemaVersionRejected is called when the version gate (§6.7) quarantines a record whose bq-schema-version header is present but not the core's supported version. version is the header value as text and queue the topic it was fetched from. The gate runs before the body is decoded; the App's quarantine error names the reserved route, and this hook additionally hands an operator the rejected version and queue directly (including for a body that is not valid JSON). It is a diagnostic hook: a panic in it is recovered and ignored, and it never changes how the record is delivered, committed or retried.
type Option ¶
type Option func(*config)
Option customizes New.
func WithBrokers ¶
WithBrokers sets the bootstrap broker list (host:port).
func WithGroupID ¶
WithGroupID sets the consumer group used for offset commits (required to consume).
func WithMaxWaitTime ¶
WithMaxWaitTime caps how long a single Pop blocks waiting for a record.
func WithOnSchemaVersionRejected ¶ added in v1.1.0
func WithOnSchemaVersionRejected(fn OnSchemaVersionRejected) Option
WithOnSchemaVersionRejected registers an optional OnSchemaVersionRejected hook (default: none).
func WithReaderFactory ¶
WithReaderFactory injects a per-queue Reader factory (a fake in tests).
func WithRetryTiers ¶ added in v1.4.0
WithRetryTiers configures the delay tiers of the retry-topic pattern (broker-bindings.md §6.4): tier n (1-based, ascending delay) is the topic `<topic>.retry.<n>`. Tiers are sorted; the set of retry topics must be declared on the broker by the operator. Without tiers, a delayed publish or a retry republish fails with ErrDelayUnsupported. Pass DefaultRetryTiers for the recommended list.
This option enables the publish side. The retry topics are drained back into the work topic by the Reinjector (§6.4), which a deployment must run for the tiers it configures.
func WithWriter ¶
WithWriter injects a Writer (a fake in tests), bypassing all Kafka wiring.
type Reader ¶
type Reader interface {
FetchMessage(ctx context.Context) (kafkago.Message, error)
CommitMessages(ctx context.Context, msgs ...kafkago.Message) error
}
Reader is the subset of *kafkago.Reader the transport uses (manual commit — no auto-commit).
type Reinjector ¶ added in v1.5.0
type Reinjector struct {
// contains filtered or unexported fields
}
Reinjector drains the retry topics `<topic>.retry.<n>` back into their work topics (broker-bindings.md §6.4). Kafka has no delayed delivery; the retry-topic pattern parks a record in a tier topic, and this consumer carries it back once its tier has elapsed. A deployment that configures WithRetryTiers MUST run one Reinjector per set of work topics, or retried and delayed records are never redelivered.
For every record of `<topic>.retry.<n>` it:
- waits until `record.timestamp + tier_ms`, clamped to [now, now+tier_ms]: the wait starts when the record was written to the retry topic, never when it was consumed, and a skewed timestamp can neither skip the delay (future-dated records wait at most one tier) nor hold the record beyond it (old records are carried at once);
- writes the value and the key, byte for byte, to the topic named by its `bq-original-topic` header (no decode, no re-encode, `attempts` untouched; an undecodable record is carried as it is and the work-topic consumer dead-letters it as poison);
- drops the `bq-delay` and `bq-original-topic` headers and keeps every other one, `traceparent` included;
- commits the retry-topic offset only after the broker acknowledged that write. A failed write is retried and never committed, so delivery is at-least-once.
kafka-go exposes no pause/resume for a group reader's partitions, so the wait is held in one worker goroutine per partition fed through a bounded buffer: the fetch loop keeps polling (the consumer stays in its group) while a record sleeps, and partitions wait independently. Records of one partition are carried in offset order.
The retry topics SHOULD be created with `message.timestamp.type=LogAppendTime`, so that `record.timestamp` is the broker's append time and a producer's clock cannot skew the wait. With the default CreateTime the producer's clock is trusted (the SDK stamps the publish time).
func (*Reinjector) Run ¶ added in v1.5.0
func (r *Reinjector) Run(ctx context.Context) error
Run consumes the retry topics until ctx is cancelled (it then returns nil) or a reader fails permanently (it returns that error). Records fetched but not yet committed when it stops are redelivered to the next run.
type Transport ¶
type Transport struct {
// contains filtered or unexported fields
}
Transport implements babelqueue.Transport over Apache Kafka. It is safe for concurrent Publish; run Pop/Ack from a single goroutine per queue (a Kafka consumer is single-threaded).
func New ¶
New builds a transport. Provide brokers + a group id (WithBrokers / WithGroupID), or inject a Writer and a reader factory (WithWriter / WithReaderFactory) for tests.
func (*Transport) Ack ¶
func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
Ack commits the reserved record's offset (process-then-commit; at-least-once).
func (*Transport) Close ¶ added in v1.2.0
Close releases what the transport created: the readers it opened and, when it built its own Writer (no WithWriter), that Writer and its private kafka-go Transport. A Writer injected with WithWriter is left open; it belongs to the caller. After Close, Pop and Ack return ErrClosed and never open a new reader; a second Close is a no-op. It returns the first error.
func (*Transport) NewReinjector ¶ added in v1.5.0
func (t *Transport) NewReinjector(workTopics ...string) (*Reinjector, error)
NewReinjector builds a Reinjector for the given work topics, using the tiers, writer, brokers and group id the transport was configured with (WithRetryTiers, WithBrokers, WithGroupID, or the injected WithWriter / WithReaderFactory). It consumes `<topic>.retry.1` ... `<topic>.retry.N` for every work topic and returns ErrDelayUnsupported when no tiers are configured. Use a dedicated group id for the re-inject process.
func (*Transport) Pop ¶
func (t *Transport) Pop(ctx context.Context, queue string, timeout time.Duration) (*babelqueue.ReceivedMessage, error)
Pop reserves the next record (FetchMessage, no auto-commit), bounded by timeout (and WithMaxWaitTime). It reconciles attempts from the authoritative bq-attempts header. Returns (nil, nil) when no record arrives before the deadline.
func (*Transport) PublishDelayed ¶ added in v1.4.0
func (t *Transport) PublishDelayed(ctx context.Context, queue, body string, delay time.Duration) error
PublishDelayed delays body on queue by publishing it to the smallest retry topic whose tier is at least delay, with `bq-delay` and `bq-original-topic` set (§6.4). It returns ErrDelayUnsupported when no tiers are configured or delay exceeds the largest tier; nothing is written then.
This only writes to the retry topic. The message reaches the work topic again once a re-inject consumer (Reinjector, or another SDK's) reads the retry topic, waits out the tier and writes the record back to `bq-original-topic`; without one it is never redelivered.
func (*Transport) PublishFrom ¶ added in v1.2.0
func (t *Transport) PublishFrom(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage) error
PublishFrom implements babelqueue.KeyedRepublisher. It has two modes:
- Restore: queue is the topic of the reserved record and body is the body Pop returned for it (the redrive tool writing an unchanged DLQ record back). The original record is re-written verbatim: its value (a null value stays null), headers and key, with a fresh timestamp so the restored copy does not age out of the DLQ early. Nothing is re-projected from the body.
- Republish: any other queue/body (a redrive to the work topic, a retry) is written like Publish (§6 header projection, created_at timestamp) and carries the key of the reserved record byte-for-byte — a null key stays null and a non-empty key is unchanged (§6.5). Only a republish onto the reserved record's own topic (the App's retry) also forwards its W3C trace headers (traceparent, tracestate); a redrive or dead-letter write to another topic copies no header, so a redrive starts a fresh trace. PublishFrom writes whatever key bytes the reserved record carries, but a record read through Pop has already lost an empty key: kafka-go's Reader delivers a zero-length key as nil, so it is written as null (see the CHANGELOG "Known limitations").
A from that carries no Kafka record (not reserved by this transport) yields a keyless record, exactly as Publish does. The key is never synthesised.
func (*Transport) PublishFromWithHeaders ¶ added in v1.6.0
func (t *Transport) PublishFromWithHeaders(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage, headers map[string]string) error
PublishFromWithHeaders is the header-carrying republish (babelqueue.HeaderRepublisher): it writes body like the republish mode of Transport.PublishFrom - §6 header projection, created_at timestamp and the key of the reserved record byte-for-byte (§6.5) - and adds the out-of-band transport headers beside the projection, with the same rules as Transport.PublishWithHeaders (a bq- contract header wins a collision, blanks are skipped). babelqueue.Redrive uses it to stamp bq-replay-bypass without losing the DLQ record's key. There is no restore mode: the headers make it a new record. A from that carries no Kafka record yields a keyless record, as PublishWithHeaders does.
func (*Transport) PublishRetry
deprecated
added in
v1.4.0
func (t *Transport) PublishRetry(ctx context.Context, from *babelqueue.ReceivedMessage, delay time.Duration) error
PublishRetry republishes the reserved record from to the smallest retry tier that covers delay. It carries the pre-1.x delay-to-tier mapping unchanged: the envelope's attempts and the `bq-attempts` header both become attempts + 1, `bq-original-topic` and `bq-delay` (= delay) are set, and the reserved record's key and W3C trace headers are carried over. The caller commits the original offset afterwards (Transport.Ack). It returns ErrDelayUnsupported when no tiers are configured or delay exceeds the largest tier.
Deprecated: the retry target of broker-bindings.md §6.5 is chosen from the attempt count, not from a delay. Use Transport.PublishRetryForAttempt.
func (*Transport) PublishRetryForAttempt ¶ added in v1.5.0
func (t *Transport) PublishRetryForAttempt(ctx context.Context, from *babelqueue.ReceivedMessage, attemptsBefore int) error
PublishRetryForAttempt republishes the reserved record from for a release/retry (§6.5). The target tier is n = min(attemptsBefore, N-1) + 1 of the N configured tiers (the last tier is kept once they run out) and the delay is that tier's own, so `bq-delay` = the tier in milliseconds. The envelope's attempts and the `bq-attempts` header both become attemptsBefore + 1, `bq-original-topic` is the reserved record's topic, the record key is carried over byte for byte and the W3C trace headers (`traceparent`, `tracestate`) of the reserved record are forwarded. The caller commits the original offset afterwards (Transport.Ack). It returns ErrDelayUnsupported when no tiers are configured; nothing is written then.
This only writes to the retry topic; the Reinjector returns the record to the work topic once the tier has elapsed. Unknown envelope keys survive the republish (ADR-0031).
func (*Transport) PublishWithHeaders ¶ added in v1.6.0
func (t *Transport) PublishWithHeaders(ctx context.Context, queue, body string, headers map[string]string) error
PublishWithHeaders implements babelqueue.HeaderPublisher: it writes body like Transport.Publish and adds the out-of-band transport headers as extra record headers (UTF-8 byte values) beside the §6 bq- projection - e.g. a W3C traceparent / tracestate for cross-hop span linkage (ADR-0028). The record value is the unchanged envelope (GR-1). A bq- contract header (and bq-delay / bq-original-topic) always wins a key collision, blank keys and blank values are skipped, and the headers are appended in key order so the record is deterministic. A nil or empty map behaves exactly like Publish.
func (*Transport) Rewind ¶ added in v1.1.0
func (t *Transport) Rewind(_ context.Context, msg *babelqueue.ReceivedMessage) error
Rewind implements babelqueue.Rewinder: it puts the consumer back at the reserved record's offset without committing anything. A Kafka commit is cumulative, so merely not acknowledging this record is not enough: committing any later record would move the group past it and lose it. A kafka-go group reader cannot seek (SetOffset is rejected when a GroupID is set), so the reader is closed and dropped; the next Pop builds a new one, which rejoins the group and resumes from the last committed offset, i.e. this record. The commit therefore never passes it.
Source Files