feat(admin): add KIP-396 list/alter offsets APIs

[DCjanus]

Summary

• Add Admin APIs for KIP-396 ListOffsets and AlterConsumerGroupOffsets to support bulk offset queries and group offset commits.
• Provide Go types aligned with KIP-396 semantics (OffsetSpec, OffsetAndMetadata).

Key changes

• Add ListOffsets and AlterConsumerGroupOffsets to ClusterAdmin and wire protocol requests.
• Add request/response types for offset specs and result structs with leader epoch support.
• Implement broker fan-out for list offsets and coordinator commit path for alter offsets.
• Add functional tests covering timestamp list offsets and admin offset commits.

Constraints / tradeoffs

• OffsetSpec is a new type instead of reusing GetOffset(int64) because GetOffset relies on magic int64 values (earliest/latest) while ListOffsets needs an explicit spec (earliest/latest/timestamp); this matches the Java AdminClient surface defined in KIP-396 and keeps room for future spec variants without breaking changes.

Notes

• Spec reference: KIP-396
• Closes #1669

[puellanivis]

I’m not sure why we need a ton of constructors, when we can simply export this value.

After all, given the functionality provided, we can already arbitrarily mutate any given OffsetSpec and access the field at will:

offspec := OffsetSpecLatest()
offspec = OffsetSpecForTimestamp(arbitraryTimestamp)
go func() {
	offspec = OffsetSpecForTimestamp(OffsetNewest)
}()
ts := offspec.Timestamp() // write-read race condition

[DCjanus]

I’m not sure why we need a ton of constructors, when we can simply export this value.

After all, given the functionality provided, we can already arbitrarily mutate any given OffsetSpec and access the field at will:

offspec := OffsetSpecLatest()
offspec = OffsetSpecForTimestamp(arbitraryTimestamp)
go func() {
	offspec = OffsetSpecForTimestamp(OffsetNewest)
}()
ts := offspec.Timestamp() // write-read race condition

Agree that we should not mix two different input styles for the same concept (as noted in #3419 (comment)).

To stay consistent with existing Sarama APIs (e.g. Client.GetOffset(topic, partition, time int64) which uses OffsetOldest/OffsetNewest), I removed OffsetSpec and made ListOffsets take int64 directly (pass OffsetOldest/OffsetNewest or a millisecond timestamp).

Done in commit f3e40c7.

[puellanivis]

var errs []error
for res := range results {
	if res.err != nil {
		errs = append(errs, res.err)
	}
	...
}

return allResults errors.Join(errs...)

[DCjanus]

var errs []error
for res := range results {
	if res.err != nil {
		errs = append(errs, res.err)
	}
	...
}

return allResults errors.Join(errs...)

Addressed in commit bce2339 by aggregating all errors with errors.Join.

[puellanivis]

How many requests are we likely to be spinning off here?

It can happen that spinning off too many goroutines will actually be performance-degrading rather than performance-enhancing.

[DCjanus]

How many requests are we likely to be spinning off here?

It can happen that spinning off too many goroutines will actually be performance-degrading rather than performance-enhancing.

Thanks for the note. The Java AdminClient doesn’t appear to impose an explicit concurrency cap either — it groups by broker and issues one in‑flight request per broker via the admin I/O loop.

Key entry point (no concurrency limiting logic):

• KafkaAdminClient#listOffsets

In practice, the Java I/O loop is equivalent to spawning goroutines in Go: both drive concurrent in‑flight requests without a per‑broker cap.

Given typical Kafka clusters are well below 1,000 brokers, I believe even 10,000 concurrent in‑flight requests should be relatively easy for modern hardware to handle.

[puellanivis]

Our current design uses map[string]map[int32]V adding this TopicPartitionID might have been nice if we had started with it, but we haven’t really. https://pkg.go.dev/github.com/IBM/sarama#AlterPartitionReassignmentsResponse

Providing two different ways to do something is likely to increase confusion over any benefits of flattening the maps.

It also reduces the ability to iterate over topics individually, without needing to then iterate over all Topic × Partition combinations and select for Topic.

[DCjanus]

Our current design uses map[string]map[int32]V adding this TopicPartitionID might have been nice if we had started with it, but we haven’t really. https://pkg.go.dev/github.com/IBM/sarama#AlterPartitionReassignmentsResponse

Providing two different ways to do something is likely to increase confusion over any benefits of flattening the maps.

It also reduces the ability to iterate over topics individually, without needing to then iterate over all Topic × Partition combinations and select for Topic.

Thanks for the point about keeping the API shape consistent. I reverted to the existing map[string]map[int32]V style so callers can iterate by topic without scanning all partitions.

This removes TopicPartitionID from ListOffsets and aligns the input/return maps with the rest of Sarama’s admin APIs.

Done in commit be65797.

[puellanivis]

I’m not sure this is an Offsets result? As it seems to be single offset result?

[DCjanus]

Renamed ListOffsetsResult to OffsetResult to reflect the single-partition result; done in commit f9150fe.

[puellanivis]

We could also isolate/scope these types into the ListOffsets right?

[DCjanus]

We could also isolate/scope these types into the ListOffsets right?

Updated in commit 243445e (scoped the helper types inside ListOffsets).

[puellanivis]

I’m not sure why we double track the broker? We’re using it as the key, and as a field in the value?

We could just for broker, req := range requests { … } below, right?

[DCjanus]

I’m not sure why we double track the broker? We’re using it as the key, and as a field in the value?

We could just for broker, req := range requests { … } below, right?

Addressed in commit ea32b6e by removing the redundant broker field and passing the broker via the map key in the loop.