Method Fetch

fetch.go:88–184

Fetch sends a fetch request to a kafka broker and returns the response. If the broker returned an invalid response with no topics, an error wrapping protocol.ErrNoTopic is returned. If the broker returned an invalid response with no partitions, an error wrapping ErrNoPartitions is returned.

(ctx context.Context, req *FetchRequest)

// If the broker returned an invalid response with no partitions, an error
// wrapping ErrNoPartitions is returned.
func (c *Client) Fetch(ctx context.Context, req *FetchRequest) (*FetchResponse, error) {
	timeout := c.timeout(ctx, math.MaxInt64)
	maxWait := req.maxWait()

	if maxWait < timeout {
		timeout = maxWait
	}

	offset := req.Offset
	switch offset {
	case FirstOffset, LastOffset:
		topic, partition := req.Topic, req.Partition

		r, err := c.ListOffsets(ctx, &ListOffsetsRequest{
			Addr: req.Addr,
			Topics: map[string][]OffsetRequest{
				topic: {{
					Partition: partition,
					Timestamp: offset,
				}},
			},
		})
		if err != nil {
			return nil, fmt.Errorf("kafka.(*Client).Fetch: %w", err)
		}

		for _, p := range r.Topics[topic] {
			if p.Partition == partition {
				if p.Error != nil {
					return nil, fmt.Errorf("kafka.(*Client).Fetch: %w", p.Error)
				}
				switch offset {
				case FirstOffset:
					offset = p.FirstOffset
				case LastOffset:
					offset = p.LastOffset
				}
				break
			}
		}
	}

	m, err := c.roundTrip(ctx, req.Addr, &fetchAPI.Request{
		ReplicaID:      -1,
		MaxWaitTime:    milliseconds(timeout),
		MinBytes:       int32(req.MinBytes),
		MaxBytes:       int32(req.MaxBytes),
		IsolationLevel: int8(req.IsolationLevel),
		SessionID:      -1,
		SessionEpoch:   -1,
		Topics: []fetchAPI.RequestTopic{{
			Topic: req.Topic,
			Partitions: []fetchAPI.RequestPartition{{
				Partition:          int32(req.Partition),
				CurrentLeaderEpoch: -1,
				FetchOffset:        offset,
				LogStartOffset:     -1,
				PartitionMaxBytes:  int32(req.MaxBytes),