kadm: change FetchOffsetsForTopics to only return requested topics by…

… default

Closes #524.

pkg/kadm/groups.go

@@ -830,6 +830,14 @@ func (cl *Client) FetchOffsets(ctx context.Context, group string) (OffsetRespons
 	return rs, nil
 }
 
+// FetchAllGroupTopics is a kadm "internal" topic name that can be used in
+// [FetchOffsetsForTopics]. By default, [FetchOffsetsForTopics] only returns
+// topics that are explicitly requested. Other topics that may be committed to
+// in the group are not returned. Using FetchAllRequestedTopics switches the
+// behavior to return the union of all committed topics and all requested
+// topics.
+const FetchAllGroupTopics = "|fetch-all-group-topics|"
+
 // FetchOffsetsForTopics is a helper function that returns the currently
 // committed offsets for the given group, as well as default -1 offsets for any
 // topic/partition that does not yet have a commit.
@@ -838,9 +846,31 @@ func (cl *Client) FetchOffsets(ctx context.Context, group string) (OffsetRespons
 // error. The returned offset responses are ready to be used or converted
 // directly to pure offsets with `Into`, and again into kgo offsets with
 // another `Into`.
+//
+// By default, this function returns offsets for only the requested topics. You
+// can use the special "topic" [FetchAllGroupTopics] to return all committed-to
+// topics in addition to all requested topics.
 func (cl *Client) FetchOffsetsForTopics(ctx context.Context, group string, topics ...string) (OffsetResponses, error) {
 	os := make(Offsets)
 
+	var all bool
+	keept := topics[:0]
+	for _, topic := range topics {
+		if topic == FetchAllGroupTopics {
+			all = true
+			continue
+		}
+		keept = append(keept, topic)
+	}
+	topics = keept
+
+	if !all && len(topics) == 0 {
+		return make(OffsetResponses), nil
+	}
+
+	// We have to request metadata to learn all partitions in all the
+	// topics. The default returned offset for all partitions is filled in
+	// to be -1.
 	if len(topics) > 0 {
 		listed, err := cl.ListTopics(ctx, topics...)
 		if err != nil {
@@ -865,11 +895,29 @@ func (cl *Client) FetchOffsetsForTopics(ctx context.Context, group string, topic
 	if err := resps.Error(); err != nil {
 		return nil, fmt.Errorf("offset fetches had a load error, first error: %w", err)
 	}
+
+	// For any topic (and any partition) we explicitly asked for, if the
+	// partition does not exist in the response, we fill the default -1
+	// from above.
 	os.Each(func(o Offset) {
 		if _, ok := resps.Lookup(o.Topic, o.Partition); !ok {
 			resps.Add(OffsetResponse{Offset: o})
 		}
 	})
+
+	// If we are not requesting all group offsets, then we strip any topic
+	// that was not explicitly requested.
+	if !all {
+		tset := make(map[string]struct{})
+		for _, t := range topics {
+			tset[t] = struct{}{}
+		}
+		for t := range resps {
+			if _, ok := tset[t]; !ok {
+				delete(resps, t)
+			}
+		}
+	}
 	return resps, nil
 }