feat(source/kafka): cold-group seek/lag, typed start-offset option, honest vendor-boundary docs

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)
 }

source/kafka/README.md

@@ -58,28 +58,47 @@ 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()`.
 
 `BlockRebalanceOnPoll` gives the engine a safe processing window: a rebalance
 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.
 

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,22 +306,27 @@ 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 {
 		return 0, fmt.Errorf("source/kafka: lag: %w", errSeekUnavailable)
 	}
 	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,13 +519,17 @@ 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
 	partition int32
 }
 
 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)
 }