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
- type OnSchemaVersionRejected
- type 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) PublishFrom(ctx context.Context, queue, body string, from *babelqueue.ReceivedMessage) 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).
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) 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) 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) 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