Function TestClientFetch

fetch_test.go:49–74

(t *testing.T)

}

func TestClientFetch(t *testing.T) {
	client, topic, shutdown := newLocalClientAndTopic()
	defer shutdown()

	records := produceRecords(t, 10, client.Addr, topic, nil)

	res, err := client.Fetch(context.Background(), &FetchRequest{
		Topic:     topic,
		Partition: 0,
		Offset:    0,
		MinBytes:  1,
		MaxBytes:  64 * 1024,
		MaxWait:   100 * time.Millisecond,
	})

	if err != nil {
		t.Fatal(err)
	}

	assertFetchResponse(t, res, &FetchResponse{
		Topic:         topic,
		Partition:     0,
		HighWatermark: 10,
		Records:       NewRecordReader(records...),
	})
}

func TestClientFetchCompressed(t *testing.T) {
	client, topic, shutdown := newLocalClientAndTopic()