Method ListOffsets

listoffset.go:79–184

ListOffsets sends an offset request to a kafka broker and returns the response.

(ctx context.Context, req *ListOffsetsRequest)

// ListOffsets sends an offset request to a kafka broker and returns the
// response.
func (c *Client) ListOffsets(ctx context.Context, req *ListOffsetsRequest) (*ListOffsetsResponse, error) {
	type topicPartition struct {
		topic     string
		partition int
	}

	partitionOffsets := make(map[topicPartition]PartitionOffsets)

	for topicName, requests := range req.Topics {
		for _, r := range requests {
			key := topicPartition{
				topic:     topicName,
				partition: r.Partition,
			}

			partition, ok := partitionOffsets[key]
			if !ok {
				partition = PartitionOffsets{
					Partition:   r.Partition,
					FirstOffset: -1,
					LastOffset:  -1,
					Offsets:     make(map[int64]time.Time),
				}
			}

			switch r.Timestamp {
			case FirstOffset:
				partition.FirstOffset = 0
			case LastOffset:
				partition.LastOffset = 0
			}

			partitionOffsets[topicPartition{
				topic:     topicName,
				partition: r.Partition,
			}] = partition
		}
	}

	topics := make([]listoffsets.RequestTopic, 0, len(req.Topics))

	for topicName, requests := range req.Topics {
		partitions := make([]listoffsets.RequestPartition, len(requests))

		for i, r := range requests {
			partitions[i] = listoffsets.RequestPartition{
				Partition:          int32(r.Partition),
				CurrentLeaderEpoch: -1,
				Timestamp:          r.Timestamp,
			}
		}

		topics = append(topics, listoffsets.RequestTopic{
			Topic:      topicName,
			Partitions: partitions,
		})
	}