Function TestClientProduceAndConsume

client_test.go:194–304

(t *testing.T)

}

func TestClientProduceAndConsume(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	// Tests a typical kafka use case, data is produced to a partition,
	// then consumed back sequentially. We use snappy compression because
	// kafka stream are often compressed, and verify that each record
	// produced is exposed to the consumer, and order is preserved.
	client, topic, shutdown := newLocalClientAndTopic()
	defer shutdown()

	epoch := time.Now()
	seed := int64(0) // deterministic
	prng := rand.New(rand.NewSource(seed))
	offset := int64(0)

	const numBatches = 100
	const recordsPerBatch = 320
	t.Logf("producing %d batches of %d records...", numBatches, recordsPerBatch)

	for i := 0; i < numBatches; i++ { // produce 100 batches
		records := make([]Record, recordsPerBatch)

		for i := range records {
			v := make([]byte, prng.Intn(999)+1)
			io.ReadFull(prng, v)
			records[i].Time = epoch
			records[i].Value = NewBytes(v)
		}

		res, err := client.Produce(ctx, &ProduceRequest{
			Topic:        topic,
			Partition:    0,
			RequiredAcks: -1,
			Records:      NewRecordReader(records...),
			Compression:  compress.Snappy,
		})
		if err != nil {
			t.Fatal(err)
		}
		if res.Error != nil {
			t.Fatal(res.Error)
		}
		if res.BaseOffset != offset {
			t.Fatalf("records were produced at an unexpected offset, want %d but got %d", offset, res.BaseOffset)
		}
		offset += int64(len(records))
	}

	prng.Seed(seed)
	offset = 0 // reset
	numFetches := 0
	numRecords := 0

	for numRecords < (numBatches * recordsPerBatch) {
		res, err := client.Fetch(ctx, &FetchRequest{
			Topic:     topic,
			Partition: 0,
			Offset:    offset,