Method ListOffsets

admin_offsets.go:42–153

ListOffsets fans out across the partition leaders to fetch offsets in parallel. Per-partition results may carry their own Err (e.g. NotLeaderForPartition, UnknownTopicOrPartition) when metadata is stale; the caller can refresh metadata via the underlying client and retry those partitions if needed.

(partitions map[string]map[int32]int64, options *ListOffsetsOptions)

// metadata via the underlying client and retry those partitions if needed. The
// retry loop here only covers transport-level failures.
func (ca *clusterAdmin) ListOffsets(partitions map[string]map[int32]int64, options *ListOffsetsOptions) (map[string]map[int32]*OffsetResult, error) {
	type topicPartition struct {
		topic     string
		partition int32
	}

	type brokerOffsetRequest struct {
		request    *OffsetRequest
		partitions []topicPartition
	}

	type brokerOffsetResult struct {
		result map[topicPartition]*OffsetResult
		err    error
	}

	if len(partitions) == 0 {
		return nil, ConfigurationError("no partitions provided")
	}

	if options == nil {
		options = &ListOffsetsOptions{}
	}

	allResults := make(map[string]map[int32]*OffsetResult, len(partitions))
	setResult := func(topic string, partition int32, result *OffsetResult) {
		if allResults[topic] == nil {
			allResults[topic] = make(map[int32]*OffsetResult)
		}
		allResults[topic][partition] = result
	}

	requests := make(map[*Broker]*brokerOffsetRequest)
	for topic, topicOffsets := range partitions {
		for partition, offsetQuery := range topicOffsets {
			broker, _, err := ca.client.LeaderAndEpoch(topic, partition)
			if err != nil {
				setResult(topic, partition, &OffsetResult{Err: err})
				continue
			}

			req := requests[broker]
			if req == nil {
				req = &brokerOffsetRequest{
					request: NewOffsetRequest(ca.conf.Version),
				}
				req.request.IsolationLevel = options.IsolationLevel
				requests[broker] = req
			}
			req.request.AddBlock(topic, partition, offsetQuery, 1)
			req.partitions = append(req.partitions, topicPartition{topic: topic, partition: partition})
		}
	}

	if len(requests) == 0 {
		return allResults, nil
	}