Function getOffsets

reader_test.go:1644–1682

(t *testing.T, config ReaderConfig)

}

func getOffsets(t *testing.T, config ReaderConfig) map[int]int64 {
	// minimal config required to lookup coordinator
	cg := ConsumerGroup{
		config: ConsumerGroupConfig{
			ID:      config.GroupID,
			Brokers: config.Brokers,
			Dialer:  config.Dialer,
		},
	}

	conn, err := cg.coordinator()
	if err != nil {
		t.Errorf("unable to connect to coordinator: %v", err)
	}
	defer conn.Close()

	offsets, err := conn.offsetFetch(offsetFetchRequestV1{
		GroupID: config.GroupID,
		Topics: []offsetFetchRequestV1Topic{{
			Topic:      config.Topic,
			Partitions: []int32{0},
		}},
	})
	if err != nil {
		t.Errorf("bad fetchOffsets: %v", err)
	}

	m := map[int]int64{}

	for _, r := range offsets.Responses {
		if r.Topic == config.Topic {
			for _, p := range r.PartitionResponses {
				m[int(p.Partition)] = p.Offset
			}
		}
	}

	return m
}

const (
	connTO     = 1 * time.Second