Function testConsumerGroupFetchOffsets

client_test.go:130–192

(t *testing.T, ctx context.Context, client *Client)

}

func testConsumerGroupFetchOffsets(t *testing.T, ctx context.Context, client *Client) {
	const totalMessages = 144
	const partitions = 12
	const msgPerPartition = totalMessages / partitions

	topic := makeTopic()
	if err := clientCreateTopic(client, topic, partitions); err != nil {
		t.Fatal(err)
	}

	groupId := makeGroupID()
	brokers := []string{"localhost:9092"}

	writer := &Writer{
		Addr:      TCP(brokers...),
		Topic:     topic,
		Balancer:  &RoundRobin{},
		BatchSize: 1,
		Transport: client.Transport,
	}
	if err := writer.WriteMessages(ctx, makeTestSequence(totalMessages)...); err != nil {
		t.Fatalf("bad write messages: %v", err)
	}
	if err := writer.Close(); err != nil {
		t.Fatalf("bad write err: %v", err)
	}

	r := NewReader(ReaderConfig{
		Brokers:  brokers,
		Topic:    topic,
		GroupID:  groupId,
		MinBytes: 1,
		MaxBytes: 10e6,
		MaxWait:  100 * time.Millisecond,
	})
	defer r.Close()

	for i := 0; i < totalMessages; i++ {
		m, err := r.FetchMessage(ctx)
		if err != nil {
			t.Fatalf("error fetching message: %s", err)
		}
		if err := r.CommitMessages(context.Background(), m); err != nil {
			t.Fatal(err)
		}
	}

	offsets, err := client.ConsumerOffsets(ctx, TopicAndGroup{GroupId: groupId, Topic: topic})
	if err != nil {
		t.Fatal(err)
	}

	if len(offsets) != partitions {
		t.Fatalf("expected %d partitions but only received offsets for %d", partitions, len(offsets))
	}

	for i := 0; i < partitions; i++ {
		committedOffset := offsets[i]