source/kafka

import "github.com/stablekernel/crucible/source/kafka"

Package kafka is a source ingress adapter that consumes records from Apache Kafka (and API-compatible brokers such as RedPanda) through the pure-Go franz-go client. It depends only on the standard library, crucible/source, and franz-go. Construct an Inlet with New, hand it to a source.Hopper, and the engine drives the consume loop, decoding, ordering, and settlement.

Ack model

Delivery is at-least-once within a live subscription: the adapter never commits an offset before its handler reports success. The franz-go client is configured with AutoCommitMarks, so only records the engine settles successfully are marked, and the marked offsets are committed on graceful drain and on rebalance. Each handler [source.Result] maps onto Kafka as follows:

Ack marks the record for commit (commit-after-process).
Nak never marks the record and redelivers it in-session: the partition is paused, re-seeked to the record’s offset so it is fetched again, then resumed; a requeue delay waits out the pause. Redelivery across restarts or rebalances rides committed offsets, so a concurrently committed higher offset can pass the nacked record — a documented best-effort divergence, not an in-session one.
Term produces the record to the configured dead-letter topic, then marks it for commit so it is not re-read — except on a transactional subscription, where Term reports ErrTermInsideTransaction and leaves the record unmarked, so it is redelivered: route poison through Begin’s transaction instead, producing to the dead-letter topic via the handed [source.Tx] so DLQ write and offset commit are atomic.
InProgress is a no-op: Kafka has no per-message ack deadline to extend.
Manual is a no-op: the handler settled the record itself through [source.Message.As] and the underlying *kgo.Client.

Cold start

A brand-new consumer group (no committed offsets) starts at the earliest retained record in each assigned partition — franz-go’s default. Build the inlet with WithStartOffset to start at the stream tail instead. Once a group has committed offsets, restarts always resume from them.

Capabilities

The [Subscription] this adapter opens satisfies several optional capability interfaces the engine discovers by type assertion, without leaking vendor types: [source.Seekable] (live offset reposition via SetOffsets), [source.ConsumerGroups] (rebalance assign/revoke/lost hooks with drain-and-commit on revoke), [source.PartitionOrdered] (per-partition order), [source.LagReporter] (end-offset lag), and [source.Transactional] (Kafka exactly-once, via a group transact session) when constructed with WithTransactional.

Vendor boundary and escape hatch

The neutral seam stays vendor-free: no franz-go type crosses the [source.Inlet], [source.Subscription], or [source.Message] surface, and the capability interfaces ([source.Seekable], [source.ConsumerGroups], …) carry only crucible types. Typed power seams do expose franz-go deliberately: WithSASL takes sasl.Mechanism values, WithBalancer takes kgo.GroupBalancer values, WithClientOptions appends raw kgo.Opt entries, and WithClient injects a pre-built *kgo.Client. Power users reach the underlying *kgo.Client through the inlet’s As method (Inlet.As) and a delivered record through [source.Message.As] with a **kgo.Record target.

Stability

Stable at v1.0.0: the adapter surface is frozen (see CHANGELOG.md for the frozen-contract terms). The typed power seams (WithSASL, WithBalancer, WithClientOptions, WithClient) may track franz-go releases.

Variables

ErrNoCommittedOffsets reports that [source.LagReporter] was called before the group committed its first offset. Without a commit there is no baseline to measure lag from; poll and settle at least one record first. Match it with errors.Is.

var ErrNoCommittedOffsets = errors.New("source/kafka: lag requires at least one committed offset")

ErrNoDLQTopic reports that a record was termed (dead-lettered) but the inlet was constructed without a dead-letter topic. Configure one with WithDLQTopic. Match it with errors.Is.

var ErrNoDLQTopic = errors.New("source/kafka: term requested but no dead-letter topic configured")

ErrNoSeedBrokers reports that New was called without any seed brokers and without an injected client. Match it with errors.Is.

var ErrNoSeedBrokers = errors.New("source/kafka: no seed brokers configured")

ErrTermInsideTransaction reports that a handler returned [source.Term] (or [source.Reject]) against a transactional subscription outside [source.Transactional.Begin]. A DLQ produce there would write outside the open EOS session and break atomicity with the consumed offset. On a transactional subscription, route poison through Begin instead: inside fn, produce the rejected record to the dead-letter topic via the handed [source.Tx] (a ProducedRecord with the DLQ topic), then return nil — the DLQ record and the input offset then commit as one atomic unit. Match it with errors.Is.

var ErrTermInsideTransaction = errors.New("source/kafka: dead-letter via term is not supported outside Begin on a transactional subscription")

type Inlet

Inlet is a franz-go-backed [source.Inlet]. It builds (or wraps) a *kgo.Client and opens a [source.Subscription] that the engine drives. One Inlet maps to one client; open a single Subscription per Inlet (the Hopper drives one).

Inlet is safe for concurrent use across its own methods, though a single Subscribe is the expected pattern.

type Inlet struct {
    // contains filtered or unexported fields
}

func New

func New(opts ...Option) (*Inlet, error)

New constructs an Inlet from opts. It does not dial: the franz-go client is built lazily on the first Subscribe (so the consumer group and topics are known), unless a client was injected with WithClient. New fails only if no seed brokers and no client were provided.

Example

ExampleNew builds an Inlet over a Kafka cluster and reports its name. In a real program the Inlet is handed to a source.Hopper, which drives the consume loop, decoding, ordering, and settlement; here we only construct it to keep the example broker-free.

package main

import (
  "context"
  "fmt"

  kafkasource "github.com/stablekernel/crucible/source/kafka"
)

func main() {
  inlet, err := kafkasource.New(
    kafkasource.WithSeedBrokers("localhost:9092"),
    kafkasource.WithClientID("orders-consumer"),
    kafkasource.WithDLQTopic("orders.DLQ"),
  )
  if err != nil {
    fmt.Println("new:", err)
    return
  }
  defer func() { _ = inlet.Close() }()

  // The inlet is a source.Inlet: hand it to a Hopper to consume.
  //   sub, _ := inlet.Subscribe(ctx, source.SubscribeConfig{Topics: []string{"orders"}, Group: "orders"})
  //   hopper.Run(ctx, sub, handler)
  _ = context.Background()

  fmt.Println("inlet ready")
}

Output

inlet ready

func (*Inlet) As

func (in *Inlet) As(target any) bool

As assigns the underlying *kgo.Client to target if target is a **kgo.Client, returning whether it did. It is the documented escape hatch for power users who must reach the franz-go client (custom admin calls, manual commits); prefer the neutral [source.Inlet] surface. It reports false before the first Subscribe builds the client.

func (*Inlet) Close

func (in *Inlet) Close() error

Close releases the inlet’s client when the inlet built it. When a client was injected with WithClient the caller owns its lifecycle and Close is a no-op for it. Close live subscriptions first; closing the inlet does not settle in-flight records.

func (in *Inlet) Subscribe(_ context.Context, cfg source.SubscribeConfig) (source.Subscription, error)

Subscribe builds (or reuses) the franz-go client for cfg and returns a live [source.Subscription]. At least one topic is required. The returned subscription owns the rebalance hooks, so it must be installed before the client begins polling; this method wires them when it builds the client.

type Option

Option configures an Inlet. Options are additive with zero-value defaults; a nil or empty value passed to a With* option leaves the default in place.

type Option func(*config)

func WithBalancer

func WithBalancer(balancers ...kgo.GroupBalancer) Option

WithBalancer sets the consumer-group partition-assignment balancer(s), in preference order, used when a subscription joins a group. The default is franz-go’s cooperative-sticky balancer. Construct balancers with kgo.CooperativeStickyBalancer, kgo.RangeBalancer, and friends. Nil entries are ignored.

func WithClient

func WithClient(client *kgo.Client) Option

WithClient injects a pre-built *kgo.Client, bypassing this package’s client construction entirely. The inlet then neither dials nor closes the client: its lifecycle is the caller’s. Use it to share a client or apply configuration this package does not expose. A nil client is ignored.

A client injected this way must already be configured to consume as a group member with AutoCommitMarks (and BlockRebalanceOnPoll) for the ack model to hold; the typed options are the supported path.

func WithClientID

func WithClientID(id string) Option

WithClientID sets the Kafka client ID the broker logs and quotas requests under. The default is franz-go’s. An empty ID is ignored.

func WithClientOptions

func WithClientOptions(opts ...kgo.Opt) Option

WithClientOptions appends raw franz-go options for tuning this package does not surface (fetch sizes, timeouts, instrumentation). They are applied after the options this package derives, so they win on conflict. It is the power-user seam; prefer the typed With* options above.

func WithDLQTopic

func WithDLQTopic(topic string) Option

WithDLQTopic sets the topic an ActionTerm result produces a rejected record to before committing it. Without a dead-letter topic a Term fails with ErrNoDLQTopic rather than silently dropping the record. An empty topic is ignored.

func WithMaxPollRecords

func WithMaxPollRecords(n int) Option

WithMaxPollRecords bounds how many records a single fetch yields, the lever the engine’s bounded in-flight window rides on. The default (0) lets franz-go decide. A value < 1 is ignored.

func WithSASL

func WithSASL(mechanisms ...sasl.Mechanism) Option

WithSASL configures one or more SASL authentication mechanisms (PLAIN, SCRAM, OAUTHBEARER, AWS_MSK_IAM) built from franz-go’s pkg/sasl helpers. The franz-go sasl.Mechanism is an interface, not a struct, so this keeps no franz-go data type in the exported surface beyond the auth abstraction itself. Nil entries are ignored.

func WithSeedBrokers

func WithSeedBrokers(brokers ...string) Option

WithSeedBrokers sets the broker addresses (host:port) the inlet dials to discover the cluster. At least one is required unless a client is injected with WithClient. Empty input is ignored.

func WithStartOffset

func WithStartOffset(o StartOffset) Option

WithStartOffset sets where a brand-new consumer group (no committed offsets) starts consuming: StartEarliest (the default) reads from the earliest retained record, StartLatest reads only records produced after joining. It maps onto franz-go’s kgo.ConsumeStartOffset(kgo.NewOffset().AtStart()|AtEnd()); partitions that already have a committed offset always resume from the commit regardless of this setting. Values outside the StartOffset constants fall back to the default.

func WithTLS

func WithTLS(cfg *tls.Config) Option

WithTLS enables TLS on broker connections using cfg. A nil cfg is ignored (connections stay plaintext); pass a non-nil &tls.Config{} for system-default TLS.

func WithTransactional

func WithTransactional(transactionalID string) Option

WithTransactional enables Kafka exactly-once semantics: the subscription satisfies [source.Transactional], fencing the records produced while processing and the consumed offset inside one producer transaction so consume-process-produce is atomic. The inlet builds a kgo.GroupTransactSession instead of a plain client, configured with the given transactional ID, read-committed fetch isolation (so it never reads another producer’s uncommitted records), and no auto-commit (the transaction commits offsets).

transactionalID must be stable and unique to this logical consumer across restarts: it is the producer fencing token Kafka uses to detect a zombie instance. Reusing one transactional ID across two live consumers of the same partitions fences one of them. An empty transactionalID is ignored and the inlet stays non-transactional.

type StartOffset

StartOffset selects where a brand-new consumer group — one with no committed offsets — begins consuming each assigned partition. Once a group has committed offsets, restarts always resume from them; this policy governs only the cold start.

type StartOffset uint8

const (
    // StartEarliest begins a cold group at the earliest retained record of each
    // partition (franz-go's default, and this package's zero value): a new
    // consumer sees the full history of its topics.
    StartEarliest StartOffset = iota
    // StartLatest begins a cold group at each partition's tail: only records
    // produced after the subscription joins are delivered.
    StartLatest
)

Generated by gomarkdoc