Method OffsetFetch

offsetfetch.go:68–125

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

(ctx context.Context, req *OffsetFetchRequest)

// OffsetFetch sends an offset fetch request to a kafka broker and returns the
// response.
func (c *Client) OffsetFetch(ctx context.Context, req *OffsetFetchRequest) (*OffsetFetchResponse, error) {

	// Kafka version 0.10.2.x and above allow null Topics map for OffsetFetch API
	// which will return the result for all topics with the desired consumer group:
	// https://kafka.apache.org/0102/protocol.html#The_Messages_OffsetFetch
	// For Kafka version below 0.10.2.x this call will result in an error
	var topics []offsetfetch.RequestTopic

	if len(req.Topics) > 0 {
		topics = make([]offsetfetch.RequestTopic, 0, len(req.Topics))

		for topicName, partitions := range req.Topics {
			indexes := make([]int32, len(partitions))

			for i, p := range partitions {
				indexes[i] = int32(p)
			}

			topics = append(topics, offsetfetch.RequestTopic{
				Name:             topicName,
				PartitionIndexes: indexes,
			})
		}
	}

	m, err := c.roundTrip(ctx, req.Addr, &offsetfetch.Request{
		GroupID: req.GroupID,
		Topics:  topics,
	})

	if err != nil {
		return nil, fmt.Errorf("kafka.(*Client).OffsetFetch: %w", err)
	}

	res := m.(*offsetfetch.Response)
	ret := &OffsetFetchResponse{
		Throttle: makeDuration(res.ThrottleTimeMs),
		Topics:   make(map[string][]OffsetFetchPartition, len(res.Topics)),
		Error:    makeError(res.ErrorCode, ""),
	}

	for _, t := range res.Topics {
		partitions := make([]OffsetFetchPartition, len(t.Partitions))

		for i, p := range t.Partitions {
			partitions[i] = OffsetFetchPartition{
				Partition:       int(p.PartitionIndex),
				CommittedOffset: p.CommittedOffset,
				Metadata:        p.Metadata,
				Error:           makeError(p.ErrorCode, ""),
			}
		}

		ret.Topics[t.Name] = partitions
	}

	return ret, nil
}