Function waitForCoordinator

conn_test.go:657–678

(t *testing.T, conn *Conn, groupID string)

}

func waitForCoordinator(t *testing.T, conn *Conn, groupID string) {
	// ensure that kafka has allocated a group coordinator.  oddly, issue doesn't
	// appear to happen if the kafka been running for a while.
	const maxAttempts = 20
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		_, err := conn.findCoordinator(findCoordinatorRequestV0{
			CoordinatorKey: groupID,
		})
		if err != nil {
			if errors.Is(err, GroupCoordinatorNotAvailable) {
				time.Sleep(250 * time.Millisecond)
				continue
			} else {
				t.Fatalf("unable to find coordinator for group: %v", err)
			}
		} else {
			return
		}
	}

	t.Fatalf("unable to connect to coordinator after %v attempts", maxAttempts)
}

func createGroup(t *testing.T, conn *Conn, groupID string) (generationID int32, memberID string, stop func()) {
	waitForCoordinator(t, conn, groupID)