From 730b579ec2b4f3f46e8344bb74aaed0fb305e0a5 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sat, 22 Aug 2026 21:41:33 -0400 Subject: [PATCH] feat(source/kafka): cold-group seek/lag, typed start-offset option, honest vendor-boundary docs P1-4: SeekToStart/SeekToEnd/SeekToTime enumerated partitions from committed offsets only, so a fresh group (nothing committed) silently no-oped. Partition discovery now falls back to a kmsg MetadataRequest for any consume topic without commits; Lag keeps the committed baseline but reports an error (exported ErrNoCommittedOffsets sentinel) instead of a misleading 0 for a cold group, and excludes never-committed partitions. P1-5: WithStartOffset(kafka.StartEarliest|StartLatest) maps onto kgo.ConsumeStartOffset for brand-new groups (default earliest = franz-go default); committed partitions always resume from their commit. Cold-start posture documented in package doc and README. P1-3: 'No franz-go type appears in an exported signature' corrected to the real boundary in kafka.go package doc and README: the neutral Inlet/ Subscription/Message seam is vendor-free; WithSASL/WithBalancer/ WithClientOptions/WithClient deliberately expose franz-go types as power seams. Core Seekable/LagReporter doc comments updated to match backend behavior. Tests: cold-group seek discovery, committed-vs-metadata partition selection, mixed-topic coverage, Lag cold-group sentinel + partial-commit counting, metadata error propagation, WithStartOffset mapping. --- source/capability.go | 16 +- source/kafka/README.md | 31 ++- source/kafka/capability.go | 176 +++++++++++---- source/kafka/capability_coldgroup_test.go | 258 ++++++++++++++++++++++ source/kafka/kafka.go | 89 +++++++- source/kafka/kafka_test.go | 42 ++++ 6 files changed, 551 insertions(+), 61 deletions(-) create mode 100644 source/kafka/capability_coldgroup_test.go diff --git a/source/capability.go b/source/capability.go index 93cd6f9..905bfbb 100644 --- a/source/capability.go +++ b/source/capability.go @@ -40,9 +40,11 @@ type Partition struct { // Seekable is a [Subscription] that can reposition its read cursor to replay or // skip ahead: the basis for replay-driven state reconstruction. Seeking takes -// effect on the next [Subscription.Next]. Satisfied by Kafka (live SetOffsets) -// and JetStream (by recreating the consumer at the target). A backend that -// cannot reposition simply does not implement it. +// effect on the next [Subscription.Next]. Satisfied by Kafka (live SetOffsets; +// partitions are enumerated from committed offsets when present, else +// discovered from broker metadata, so seeking works before the group's first +// commit) and JetStream (by recreating the consumer at the target). A backend +// that cannot reposition simply does not implement it. type Seekable interface { // SeekToTime repositions delivery to the first message at or after t. SeekToTime(ctx context.Context, t time.Time) error @@ -210,9 +212,13 @@ type Deduper interface { // LagReporter is a [Subscription] that can report how far behind the tail it is, // the headline health signal for a consumer. The [Hopper] feeds it into a lag // gauge when present. Satisfied by backends that expose a high-water mark -// (Kafka end offsets, JetStream pending counts). +// (Kafka end offsets, JetStream pending counts). Backends measure from the +// committed position; Kafka reports an error until the group's first offset +// commit, and excludes never-committed partitions from the count. type LagReporter interface { // Lag returns the number of unconsumed messages between the committed position - // and the stream tail, across all assigned partitions/subjects. + // and the stream tail, across all assigned partitions/subjects. A backend + // with no committed position yet reports an error rather than a misleading + // zero. Lag(ctx context.Context) (int64, error) } diff --git a/source/kafka/README.md b/source/kafka/README.md index 36ff876..2346959 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -58,17 +58,24 @@ divergence from JetStream's native nak semantics. ## Capabilities The subscription satisfies these optional `source` capability interfaces, -discovered by the engine via type assertion — no franz-go type leaks into the -exported API: +discovered by the engine via type assertion. The neutral seam stays +vendor-free — no franz-go type crosses the `source.Inlet` / `Subscription` / +`Message` surface; the typed option and escape-hatch seams below deliberately +do expose it: - `Seekable` — live offset reposition via `SetOffsets` (and `ListOffsets` for - time-based seeks), the basis for replay. + time-based seeks), the basis for replay. Partitions are enumerated from the + group's committed offsets when present, else discovered from broker metadata, + so seeking works before the first commit. - `ConsumerGroups` — `GroupID` plus assign/revoke hooks; the adapter drain-and-commits marked offsets on a graceful revoke and skips the commit on an ungraceful loss. - `PartitionOrdered` — per-partition order, the guarantee the Hopper keys its ordered lanes on (`PartitionKey()` is `"topic/partition"`). -- `LagReporter` — end-offset minus committed offset across assigned partitions. +- `LagReporter` — end offset minus committed offset across committed + partitions. Before the group's first commit there is no baseline: `Lag` + reports an error (`ErrNoCommittedOffsets`, matchable with `errors.Is`) + rather than a misleading zero. - `Transactional` — Kafka exactly-once consume-process-produce, available when the inlet is built with `WithTransactional()`. @@ -76,10 +83,22 @@ exported API: cannot move partitions mid-batch; the subscription releases the rebalance only between fetches. +## Cold start (initial start offset) + +A brand-new consumer group — one with no committed offsets — starts at the +**earliest retained record** of each assigned partition (franz-go's default). +Build the inlet with `WithStartOffset(kafka.StartLatest)` to skip backlog and +consume only records produced after joining instead. The option maps onto +`kgo.ConsumeStartOffset`; partitions that already have a committed offset +always resume from the commit regardless of this setting. + ## Vendor escape hatch -No franz-go type appears in an exported signature. Reach the underlying -`*kgo.Client` through `Inlet.As(**kgo.Client)`, and a delivered record through +The neutral surface carries no vendor types. Typed power seams 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`. Reach that client through +`Inlet.As(**kgo.Client)` and a delivered record through `source.Message.As(**kgo.Record)`. The client lifecycle is the inlet's unless one is injected with `WithClient`, in which case it is the caller's. diff --git a/source/kafka/capability.go b/source/kafka/capability.go index 1b1b2e8..374a344 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -6,6 +6,8 @@ import ( "context" "errors" "fmt" + "maps" + "slices" "time" "github.com/twmb/franz-go/pkg/kgo" @@ -58,13 +60,19 @@ func (s *subscription) asRequester() (requester, bool) { // at or after t, taking effect on the next [Subscription.Next]. It resolves the // timestamp to per-partition offsets with a ListOffsets request, then applies // them with SetOffsets — the live-reposition path a group consumer supports -// without being recreated. +// without being recreated. Partitions are enumerated from committed offsets +// when present, else discovered from broker metadata, so a cold group (nothing +// committed yet) still seeks across its consume topics. func (s *subscription) SeekToTime(ctx context.Context, t time.Time) error { r, ok := s.asRequester() if !ok { return fmt.Errorf("source/kafka: seek to time: %w", errSeekUnavailable) } - offsets, err := listOffsets(ctx, r, s.assignedTopics(r), t.UnixMilli()) + parts, err := assignedPartitions(ctx, r) + if err != nil { + return fmt.Errorf("source/kafka: seek to time: %w", err) + } + offsets, err := listOffsets(ctx, r, parts, t.UnixMilli()) if err != nil { return fmt.Errorf("source/kafka: seek to time: %w", err) } @@ -88,50 +96,130 @@ func (s *subscription) SeekToCursor(_ context.Context, c source.Cursor) error { // SeekToStart repositions every assigned partition to its earliest retained // record (logical offset -2). -func (s *subscription) SeekToStart(_ context.Context) error { - return s.seekLogical(-2) +func (s *subscription) SeekToStart(ctx context.Context) error { + return s.seekLogical(ctx, -2) } // SeekToEnd repositions every assigned partition to its tail (logical offset // -1), skipping the backlog so only records produced after the seek are // delivered. -func (s *subscription) SeekToEnd(_ context.Context) error { - return s.seekLogical(-1) +func (s *subscription) SeekToEnd(ctx context.Context) error { + return s.seekLogical(ctx, -1) } // seekLogical applies a Kafka logical offset (-2 earliest, -1 latest) to every -// currently-assigned partition via SetOffsets. -func (s *subscription) seekLogical(logical int64) error { +// currently-assigned partition via SetOffsets. Partitions are enumerated from +// committed offsets when present, else discovered from broker metadata for the +// consume topics — so a cold group (nothing committed) still seeks instead of +// silently no-oping. +func (s *subscription) seekLogical(ctx context.Context, logical int64) error { r, ok := s.asRequester() if !ok { return fmt.Errorf("source/kafka: seek: %w", errSeekUnavailable) } + parts, err := assignedPartitions(ctx, r) + if err != nil { + return fmt.Errorf("source/kafka: seek: %w", err) + } set := map[string]map[int32]kgo.EpochOffset{} - for topic, parts := range r.CommittedOffsets() { + for topic, ps := range parts { set[topic] = map[int32]kgo.EpochOffset{} - for p := range parts { + for _, p := range ps { set[topic][p] = kgo.EpochOffset{Offset: logical, Epoch: -1} } } - if len(set) == 0 { - return nil - } s.client.SetOffsets(set) return nil } -// assignedTopics reports the topics the consumer is currently assigned, the set -// SeekToTime resolves offsets across. -func (s *subscription) assignedTopics(r requester) []string { +// assignedPartitions enumerates the partitions the consumer works across: +// committed offsets where present per topic, broker metadata discovery for any +// consume topic without commits (a cold group), and the consume topics +// themselves when nothing at all has been committed. +func assignedPartitions(ctx context.Context, r requester) (map[string][]int32, error) { committed := r.CommittedOffsets() - if len(committed) > 0 { - topics := make([]string, 0, len(committed)) - for t := range committed { - topics = append(topics, t) + consume := r.GetConsumeTopics() + if len(committed) == 0 && len(consume) == 0 { + return map[string][]int32{}, nil + } + topics := make([]string, 0, len(committed)+len(consume)) + seen := map[string]bool{} + add := func(ts []string) { + for _, t := range ts { + if t != "" && !seen[t] { + seen[t] = true + topics = append(topics, t) + } + } + } + for t := range committed { + add([]string{t}) + } + add(consume) + + out := make(map[string][]int32, len(topics)) + var discover []string + for _, t := range topics { + for p := range committed[t] { + out[t] = append(out[t], p) + } + if len(out[t]) == 0 { + discover = append(discover, t) + } + } + if len(discover) > 0 { + found, err := topicPartitions(ctx, r, discover) + if err != nil { + return nil, err + } + for t, ps := range found { + out[t] = append(out[t], ps...) + } + } + for _, t := range topics { + if len(out[t]) == 0 { + delete(out, t) + } + } + return out, nil +} + +// topicPartitions discovers a topic's partition ids from broker metadata via a +// MetadataRequest — the fallback that lets seek/lag act before the group's +// first commit. A per-topic or per-partition error code is returned as an +// error. +func topicPartitions(ctx context.Context, r requester, topics []string) (map[string][]int32, error) { + req := kmsg.NewPtrMetadataRequest() + for _, t := range topics { + rt := kmsg.NewMetadataRequestTopic() + rt.Topic = kmsg.StringPtr(t) + req.Topics = append(req.Topics, rt) + } + resp, err := r.Request(ctx, req) + if err != nil { + return nil, err + } + mr, ok := resp.(*kmsg.MetadataResponse) + if !ok { + return nil, fmt.Errorf("unexpected response type %T", resp) + } + out := make(map[string][]int32, len(topics)) + for _, t := range mr.Topics { + name := "" + if t.Topic != nil { + name = *t.Topic + } + if t.ErrorCode != 0 { + return nil, &kmsgError{code: t.ErrorCode, topic: name} + } + for _, p := range t.Partitions { + if p.ErrorCode != 0 { + return nil, &kmsgError{code: p.ErrorCode, topic: name, partition: p.Partition} + } + out[name] = append(out[name], p.Partition) } - return topics } - return r.GetConsumeTopics() + return out, nil } // --- ConsumerGroups --------------------------------------------------------- @@ -218,8 +306,11 @@ func (s *subscription) PartitionOrdered() {} // --- LagReporter ------------------------------------------------------------ // Lag reports the number of unconsumed records between the committed position -// and the stream tail across all assigned partitions. It resolves the tail with -// a ListOffsets request (timestamp -1) and subtracts the committed offsets. +// and the stream tail across all partitions with a committed offset. It +// resolves the tail with a ListOffsets request (timestamp -1) and subtracts +// the committed offsets. Partitions with no commit yet are excluded — without +// a commit there is no baseline to measure from — and a cold group (no commits +// at all) reports [ErrNoCommittedOffsets] rather than a misleading 0. func (s *subscription) Lag(ctx context.Context) (int64, error) { r, ok := s.asRequester() if !ok { @@ -227,13 +318,15 @@ func (s *subscription) Lag(ctx context.Context) (int64, error) { } committed := r.CommittedOffsets() if len(committed) == 0 { - return 0, nil + return 0, fmt.Errorf("source/kafka: lag: %w", ErrNoCommittedOffsets) } - topics := make([]string, 0, len(committed)) - for t := range committed { - topics = append(topics, t) + parts := map[string][]int32{} + for t, ps := range committed { + for p := range ps { + parts[t] = append(parts[t], p) + } } - ends, err := listOffsets(ctx, r, topics, -1) + ends, err := listOffsets(ctx, r, parts, -1) if err != nil { return 0, fmt.Errorf("source/kafka: lag: %w", err) } @@ -376,19 +469,18 @@ func (s *subscription) Begin(ctx context.Context, m source.Message, fn func(ctx return nil } -// listOffsets resolves per-partition offsets for the given topics at timestamp -// ts (millis; -1 latest, -2 earliest) via a ListOffsets request, returning the -// EpochOffset map SetOffsets consumes. Partitions are discovered from the -// requester's committed offsets so the request targets exactly what the -// consumer holds. -func listOffsets(ctx context.Context, r requester, topics []string, ts int64) (map[string]map[int32]kgo.EpochOffset, error) { - committed := r.CommittedOffsets() +// listOffsets resolves per-partition offsets for the given topic→partitions +// map at timestamp ts (millis; -1 latest, -2 earliest) via a ListOffsets +// request, returning the EpochOffset map SetOffsets consumes. The caller +// enumerates partitions (assignedPartitions), so a cold group's discovered +// partitions are included. +func listOffsets(ctx context.Context, r requester, parts map[string][]int32, ts int64) (map[string]map[int32]kgo.EpochOffset, error) { req := kmsg.NewPtrListOffsetsRequest() req.ReplicaID = -1 - for _, topic := range topics { + for _, topic := range slices.Sorted(maps.Keys(parts)) { rt := kmsg.NewListOffsetsRequestTopic() rt.Topic = topic - for p := range committed[topic] { + for _, p := range parts[topic] { rp := kmsg.NewListOffsetsRequestTopicPartition() rp.Partition = p rp.Timestamp = ts @@ -427,7 +519,8 @@ func listOffsets(ctx context.Context, r requester, topics []string, ts int64) (m return out, nil } -// kmsgError reports a per-partition error code from a ListOffsets response. +// kmsgError reports a per-topic or per-partition error code from a Kafka +// protocol response (ListOffsets, Metadata). type kmsgError struct { code int16 topic string @@ -435,5 +528,8 @@ type kmsgError struct { } func (e *kmsgError) Error() string { - return fmt.Sprintf("list offsets %s[%d]: kafka error code %d", e.topic, e.partition, e.code) + if e.topic == "" { + return fmt.Sprintf("kafka error code %d", e.code) + } + return fmt.Sprintf("kafka error code %d on %s[%d]", e.code, e.topic, e.partition) } diff --git a/source/kafka/capability_coldgroup_test.go b/source/kafka/capability_coldgroup_test.go new file mode 100644 index 0000000..9f53455 --- /dev/null +++ b/source/kafka/capability_coldgroup_test.go @@ -0,0 +1,258 @@ +// SPDX-License-Identifier: Apache-2.0 + +package kafka + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/twmb/franz-go/pkg/kgo" + "github.com/twmb/franz-go/pkg/kmsg" + + "github.com/stablekernel/crucible/source" +) + +// metaReqPoller embeds the fakePoller and dispatches Request by protocol type: +// a *kmsg.MetadataRequest is answered with metaResp and a +// *kmsg.ListOffsetsRequest with listResp, so the cold-group discovery path +// (metadata first, then list offsets) is exercised without a broker. Every +// request is recorded in requests for type-count assertions. +type metaReqPoller struct { + fakePoller + committedMap map[string]map[int32]kgo.EpochOffset + consumeTopics []string + metaResp *kmsg.MetadataResponse + listResp *kmsg.ListOffsetsResponse + requests []kmsg.Request +} + +func (m *metaReqPoller) Request(_ context.Context, req kmsg.Request) (kmsg.Response, error) { + m.requests = append(m.requests, req) + switch req.(type) { + case *kmsg.MetadataRequest: + if m.metaResp == nil { + return nil, errors.New("unexpected metadata request") + } + return m.metaResp, nil + case *kmsg.ListOffsetsRequest: + if m.listResp == nil { + return nil, errors.New("unexpected list offsets request") + } + return m.listResp, nil + default: + return nil, errors.New("unexpected request type") + } +} + +func (m *metaReqPoller) CommittedOffsets() map[string]map[int32]kgo.EpochOffset { + return m.committedMap +} + +func (m *metaReqPoller) GetConsumeTopics() []string { return m.consumeTopics } + +// countRequests tallies recorded requests by concrete type. +func countRequests(reqs []kmsg.Request) (meta, list int) { + for _, r := range reqs { + switch r.(type) { + case *kmsg.MetadataRequest: + meta++ + case *kmsg.ListOffsetsRequest: + list++ + } + } + return meta, list +} + +// metadataOrders builds a MetadataResponse declaring one topic with the given +// partition ids. +func metadataOrders(topic string, partitions ...int32) *kmsg.MetadataResponse { + ps := make([]kmsg.MetadataResponseTopicPartition, 0, len(partitions)) + for _, p := range partitions { + ps = append(ps, kmsg.MetadataResponseTopicPartition{Partition: p}) + } + return &kmsg.MetadataResponse{ + Topics: []kmsg.MetadataResponseTopic{{ + Topic: kmsg.StringPtr(topic), + Partitions: ps, + }}, + } +} + +// TestSeekToStartColdGroupDiscoversPartitions pins the P1-4 fix: a group with +// nothing committed discovers its partitions from broker metadata instead of +// silently no-oping. +func TestSeekToStartColdGroupDiscoversPartitions(t *testing.T) { + t.Parallel() + + rp := &metaReqPoller{ + consumeTopics: []string{"orders"}, + metaResp: metadataOrders("orders", 0, 1, 2), + } + sub := newSub(rp) + + if err := sub.SeekToStart(context.Background()); err != nil { + t.Fatalf("SeekToStart() error = %v", err) + } + if len(rp.setOffsets) != 1 { + t.Fatalf("setOffsets calls = %d, want 1", len(rp.setOffsets)) + } + set := rp.setOffsets[0]["orders"] + if len(set) != 3 { + t.Fatalf("partitions seeked = %d, want 3", len(set)) + } + for _, p := range []int32{0, 1, 2} { + got := set[p] + if got.Offset != -2 || got.Epoch != -1 { + t.Errorf("partition %d = %+v, want offset -2 epoch -1", p, got) + } + } + meta, list := countRequests(rp.requests) + if meta != 1 || list != 0 { + t.Errorf("requests = (metadata %d, list offsets %d), want (1, 0)", meta, list) + } +} + +// TestSeekToEndUsesCommittedWhenPresent proves the committed-offsets fast path: +// no broker requests at all when every assigned topic has commits. +func TestSeekToEndUsesCommittedWhenPresent(t *testing.T) { + t.Parallel() + + rp := &metaReqPoller{committedMap: map[string]map[int32]kgo.EpochOffset{ + "orders": {0: {Offset: 5}, 1: {Offset: 9}}, + }} + sub := newSub(rp) + + if err := sub.SeekToEnd(context.Background()); err != nil { + t.Fatalf("SeekToEnd() error = %v", err) + } + if len(rp.setOffsets) != 1 { + t.Fatalf("setOffsets calls = %d, want 1", len(rp.setOffsets)) + } + set := rp.setOffsets[0]["orders"] + for _, p := range []int32{0, 1} { + if set[p].Offset != -1 || set[p].Epoch != -1 { + t.Errorf("partition %d = %+v, want offset -1 epoch -1", p, set[p]) + } + } + if meta, list := countRequests(rp.requests); meta != 0 || list != 0 { + t.Errorf("requests = (metadata %d, list offsets %d), want none: committed partitions must be reused", meta, list) + } +} + +// TestSeekToTimeColdGroupMixedTopics covers the mixed case: one topic with a +// commit (reuse its committed partition set) and one cold topic (discover from +// metadata), then resolve timestamps across both via ListOffsets. +func TestSeekToTimeColdGroupMixedTopics(t *testing.T) { + t.Parallel() + + rp := &metaReqPoller{ + committedMap: map[string]map[int32]kgo.EpochOffset{"a": {0: {Offset: 3}}}, + consumeTopics: []string{"a", "b"}, + metaResp: &kmsg.MetadataResponse{Topics: []kmsg.MetadataResponseTopic{ + {Topic: kmsg.StringPtr("a"), Partitions: []kmsg.MetadataResponseTopicPartition{{Partition: 0}}}, + {Topic: kmsg.StringPtr("b"), Partitions: []kmsg.MetadataResponseTopicPartition{{Partition: 0}, {Partition: 1}}}, + }}, + listResp: &kmsg.ListOffsetsResponse{Topics: []kmsg.ListOffsetsResponseTopic{ + {Topic: "a", Partitions: []kmsg.ListOffsetsResponseTopicPartition{{Partition: 0, Offset: 10}}}, + {Topic: "b", Partitions: []kmsg.ListOffsetsResponseTopicPartition{ + {Partition: 0, Offset: 20}, {Partition: 1, Offset: 30}, + }}, + }}, + } + sub := newSub(rp) + + if err := sub.SeekToTime(context.Background(), time.Unix(1000, 0)); err != nil { + t.Fatalf("SeekToTime() error = %v", err) + } + if len(rp.setOffsets) != 1 { + t.Fatalf("setOffsets calls = %d, want 1", len(rp.setOffsets)) + } + set := rp.setOffsets[0] + if got := set["a"][0].Offset; got != 10 { + t.Errorf("a/0 offset = %d, want 10", got) + } + if got := set["b"][0].Offset; got != 20 { + t.Errorf("b/0 offset = %d, want 20", got) + } + if got := set["b"][1].Offset; got != 30 { + t.Errorf("b/1 offset = %d, want 30", got) + } + if meta, list := countRequests(rp.requests); meta != 1 || list != 1 { + t.Errorf("requests = (metadata %d, list offsets %d), want (1, 1)", meta, list) + } +} + +// TestLagColdGroupErrorsWithSentinel pins the Lag half of P1-4: before the +// first commit there is no baseline, so Lag reports ErrNoCommittedOffsets +// rather than a misleading zero. +func TestLagColdGroupErrorsWithSentinel(t *testing.T) { + t.Parallel() + + rp := &metaReqPoller{consumeTopics: []string{"orders"}} + sub := newSub(rp) + + _, err := sub.Lag(context.Background()) + if !errors.Is(err, ErrNoCommittedOffsets) { + t.Fatalf("Lag() error = %v, want errors.Is ErrNoCommittedOffsets", err) + } +} + +// TestLagPartialCommitsCountOnlyCommitted verifies lag counts only partitions +// with a committed baseline; an uncommitted consume topic contributes nothing +// and triggers no metadata discovery on the lag path. +func TestLagPartialCommitsCountOnlyCommitted(t *testing.T) { + t.Parallel() + + rp := &metaReqPoller{ + committedMap: map[string]map[int32]kgo.EpochOffset{"a": {0: {Offset: 10}}}, + consumeTopics: []string{"a", "b"}, + listResp: &kmsg.ListOffsetsResponse{Topics: []kmsg.ListOffsetsResponseTopic{ + {Topic: "a", Partitions: []kmsg.ListOffsetsResponseTopicPartition{{Partition: 0, Offset: 15}}}, + }}, + } + sub := newSub(rp) + + lag, err := sub.Lag(context.Background()) + if err != nil { + t.Fatalf("Lag() error = %v", err) + } + if lag != 5 { + t.Errorf("lag = %d, want 5", lag) + } + if meta, list := countRequests(rp.requests); meta != 0 || list != 1 { + t.Errorf("requests = (metadata %d, list offsets %d), want (0, 1)", meta, list) + } +} + +// TestTopicPartitionsErrorPropagates proves a metadata-level failure (unknown +// topic, error code 3) surfaces from SeekToStart instead of being swallowed. +func TestTopicPartitionsErrorPropagates(t *testing.T) { + t.Parallel() + + rp := &metaReqPoller{ + consumeTopics: []string{"missing"}, + metaResp: &kmsg.MetadataResponse{Topics: []kmsg.MetadataResponseTopic{{ + Topic: kmsg.StringPtr("missing"), + ErrorCode: 3, + }}}, + } + sub := newSub(rp) + + err := sub.SeekToStart(context.Background()) + if err == nil { + t.Fatal("SeekToStart() = nil, want a metadata error") + } + var ke *kmsgError + if !errors.As(err, &ke) || ke.code != 3 { + t.Fatalf("error = %v, want kmsgError code 3", err) + } +} + +// Compile-time reminder that Seekable/LagReporter stay satisfied by the +// subscription while the cold-group paths evolve. +var ( + _ source.Seekable = (*subscription)(nil) + _ source.LagReporter = (*subscription)(nil) +) diff --git a/source/kafka/kafka.go b/source/kafka/kafka.go index f272e1d..6361fac 100644 --- a/source/kafka/kafka.go +++ b/source/kafka/kafka.go @@ -8,11 +8,12 @@ // // # Ack model // -// Delivery is at-least-once: 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: +// 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 @@ -27,6 +28,13 @@ // - 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 @@ -38,12 +46,17 @@ // exactly-once, via a group transact session) when constructed with // [WithTransactional]. // -// # Vendor escape hatch +// # Vendor boundary and escape hatch // -// No franz-go type appears in an exported signature. A power user who must -// drop to the driver reaches 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. +// 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 // @@ -71,6 +84,12 @@ var ErrNoSeedBrokers = errors.New("source/kafka: no seed brokers configured") // [WithDLQTopic]. Match it with errors.Is. var ErrNoDLQTopic = errors.New("source/kafka: term requested but no dead-letter topic configured") +// 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") + // errTransactionalSingleSubscribe reports a second [Inlet.Subscribe] on a // transactional inlet. The exactly-once session backing a transactional inlet // fences a single consumer, so only one subscription per transactional inlet is @@ -92,6 +111,42 @@ type config struct { transactID string extraOpts []kgo.Opt client *kgo.Client + + startOffset StartOffset + startOffsetSet bool +} + +// 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 +) + +// 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 WithStartOffset(o StartOffset) Option { + return func(c *config) { + switch o { + case StartEarliest, StartLatest: + c.startOffset = o + c.startOffsetSet = true + } + } } // Option configures an [Inlet]. Options are additive with zero-value defaults; @@ -309,9 +364,22 @@ func (in *Inlet) consumeOpts(sc source.SubscribeConfig) []kgo.Opt { if len(in.cfg.balancers) > 0 { opts = append(opts, kgo.Balancers(in.cfg.balancers...)) } + opts = in.appendStartOffset(opts) return opts } +// appendStartOffset appends the typed cold-start policy onto an option set. +func (in *Inlet) appendStartOffset(opts []kgo.Opt) []kgo.Opt { + if !in.cfg.startOffsetSet { + return opts + } + off := kgo.NewOffset().AtStart() + if in.cfg.startOffset == StartLatest { + off = kgo.NewOffset().AtEnd() + } + return append(opts, kgo.ConsumeStartOffset(off)) +} + // transactOpts assembles the franz-go options for an EOS group transact session: // the base group options without auto-commit (the transaction commits offsets), // the transactional ID that fences a zombie producer, and read-committed fetch @@ -335,6 +403,7 @@ func (in *Inlet) transactOpts(sc source.SubscribeConfig) []kgo.Opt { if len(in.cfg.balancers) > 0 { opts = append(opts, kgo.Balancers(in.cfg.balancers...)) } + opts = in.appendStartOffset(opts) return opts } diff --git a/source/kafka/kafka_test.go b/source/kafka/kafka_test.go index 31391b1..4ce5ef3 100644 --- a/source/kafka/kafka_test.go +++ b/source/kafka/kafka_test.go @@ -416,3 +416,45 @@ func (foreignMessage) Subject() string { return "other" } func (foreignMessage) PartitionKey() string { return "" } func (foreignMessage) Cursor() source.Cursor { return nil } func (foreignMessage) As(any) bool { return false } + +// TestWithStartOffsetOptionConfigAndMapping pins the typed cold-start option: +// valid constants record the policy, unknown values are ignored, and +// appendStartOffset appends exactly one franz-go option when (and only when) +// the policy was set. +func TestWithStartOffsetOptionConfigAndMapping(t *testing.T) { + t.Parallel() + + // StartLatest records the policy and appends exactly one option. + cLatest := &config{} + WithStartOffset(StartLatest)(cLatest) + if !cLatest.startOffsetSet || cLatest.startOffset != StartLatest { + t.Fatalf("config = (set %v, offset %v), want (true, StartLatest)", cLatest.startOffsetSet, cLatest.startOffset) + } + inLatest := &Inlet{cfg: *cLatest} + if got := len(inLatest.appendStartOffset(nil)); got != 1 { + t.Errorf("appendStartOffset(StartLatest) appended %d opts, want 1", got) + } + + // StartEarliest (the zero value) is also an explicit, recorded policy. + cEarliest := &config{} + WithStartOffset(StartEarliest)(cEarliest) + if !cEarliest.startOffsetSet || cEarliest.startOffset != StartEarliest { + t.Fatalf("config = (set %v, offset %v), want (true, StartEarliest)", cEarliest.startOffsetSet, cEarliest.startOffset) + } + inEarliest := &Inlet{cfg: *cEarliest} + if got := len(inEarliest.appendStartOffset(nil)); got != 1 { + t.Errorf("appendStartOffset(StartEarliest) appended %d opts, want 1", got) + } + + // An out-of-range value is ignored: the default (earliest) stays in force + // and no franz-go option is appended. + cUnknown := &config{} + WithStartOffset(StartOffset(99))(cUnknown) + if cUnknown.startOffsetSet { + t.Error("startOffsetSet = true for unknown StartOffset, want false") + } + inUnknown := &Inlet{cfg: *cUnknown} + if got := len(inUnknown.appendStartOffset(nil)); got != 0 { + t.Errorf("appendStartOffset(unset) appended %d opts, want 0", got) + } +}