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 Transport
- func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) error
- func (t *Transport) Close() 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) PublishRetry(ctx context.Context, from *babelqueue.ReceivedMessage, delay time.Duration) error
- func (t *Transport) Rewind(_ context.Context, msg *babelqueue.ReceivedMessage) error
- type Writer
Constants ¶
This section is empty.
Variables ¶
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 i (1-based, ascending delay) is the topic `<topic>.retry.<i>`. 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.
This option only enables the publish side. Draining the retry topics back into the work topic after `record.timestamp + tier` (the retry-topic consumer of §6.4) is not part of this transport.
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 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) 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.
WARNING: this only writes to the retry topic. A message published here is delivered to the work topic again only if a re-inject consumer reads the retry topic, waits out `bq-delay` and writes the record back to the `bq-original-topic` work topic. The Go SDK does not provide that consumer yet, so the caller must run one (or another SDK's) or the message 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). 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) PublishRetry ¶ 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 a retry topic for a release/retry (§6.5): 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 is 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.
WARNING: like Transport.PublishDelayed, this only writes to the retry topic. The record returns to the work topic only if a re-inject consumer (retry topic -> work topic, honouring `bq-delay`) is running; the Go SDK does not provide one yet, so without it the retried message is not redelivered.
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