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 ¶
- type OnSchemaVersionRejected
- type Option
- type Reader
- type Transport
- func (t *Transport) Ack(ctx context.Context, msg *babelqueue.ReceivedMessage) 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) Rewind(_ context.Context, msg *babelqueue.ReceivedMessage) error
- type Writer
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
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 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) 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) 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