Function TestOffsetFetchRequestWithOneTopic

offsetfetch_test.go:112–167

(t *testing.T)

}

func TestOffsetFetchRequestWithOneTopic(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	topic1 := makeTopic()
	defer deleteTopic(t, topic1)
	topic2 := makeTopic()
	defer deleteTopic(t, topic2)
	consumeGroup := makeGroupID()
	numMsgs := 50
	defer cancel()
	r1 := NewReader(ReaderConfig{
		Brokers:  []string{"localhost:9092"},
		Topic:    topic1,
		GroupID:  consumeGroup,
		MinBytes: 1,
		MaxBytes: 100,
		MaxWait:  100 * time.Millisecond,
	})
	defer r1.Close()
	prepareReader(t, ctx, r1, makeTestSequence(numMsgs)...)
	r2 := NewReader(ReaderConfig{
		Brokers:  []string{"localhost:9092"},
		Topic:    topic2,
		GroupID:  consumeGroup,
		MinBytes: 1,
		MaxBytes: 100,
		MaxWait:  100 * time.Millisecond,
	})
	defer r2.Close()
	prepareReader(t, ctx, r2, makeTestSequence(numMsgs)...)

	for i := 0; i < numMsgs; i++ {
		if _, err := r1.ReadMessage(ctx); err != nil {
			t.Fatal(err)
		}
	}
	for i := 0; i < numMsgs; i++ {
		if _, err := r2.ReadMessage(ctx); err != nil {
			t.Fatal(err)
		}
	}

	client := Client{Addr: TCP("localhost:9092")}
	topicOffsets, err := client.OffsetFetch(ctx, &OffsetFetchRequest{GroupID: consumeGroup, Topics: map[string][]int{
		topic1: {0},
	}})

	if err != nil {
		t.Error(err)
		t.FailNow()
	}

	if len(topicOffsets.Topics) != 1 {
		t.Error(err)
		t.FailNow()
	}
}