Function testConnReadBatchWithMaxWait

conn_test.go:608–655

(t *testing.T, conn *Conn)

}

func testConnReadBatchWithMaxWait(t *testing.T, conn *Conn) {
	if _, err := conn.WriteMessages(makeTestSequence(10)...); err != nil {
		t.Fatal(err)
	}

	const maxBytes = 10e6 // 10 MB

	value := make([]byte, 10e3) // 10 KB

	cfg := ReadBatchConfig{
		MinBytes: maxBytes, // use max for both so that we hit max wait time
		MaxBytes: maxBytes,
		MaxWait:  500 * time.Millisecond,
	}

	// set aa read deadline so the batch will succeed.
	conn.SetDeadline(time.Now().Add(time.Second))
	batch := conn.ReadBatchWith(cfg)

	for i := 0; i < 10; i++ {
		_, err := batch.Read(value)
		if err != nil {
			if err = batch.Close(); err != nil {
				t.Fatalf("error trying to read batch message: %s", err)
			}
		}

		if batch.HighWaterMark() != 10 {
			t.Fatal("expected highest offset (watermark) to be 10")
		}
	}

	batch.Close()

	// reset the offset and  ensure that the conn deadline takes precedence over
	// the max wait
	conn.Seek(0, SeekAbsolute)
	conn.SetDeadline(time.Now().Add(50 * time.Millisecond))
	batch = conn.ReadBatchWith(cfg)
	var netErr net.Error
	if err := batch.Err(); err == nil {
		t.Fatal("should have timed out, but got no error")
	} else if errors.As(err, &netErr) {
		if !netErr.Timeout() {
			t.Fatalf("should have timed out, but got: %v", err)
		}
	}
}

func waitForCoordinator(t *testing.T, conn *Conn, groupID string) {
	// ensure that kafka has allocated a group coordinator.  oddly, issue doesn't