Function TestClientCheckBrokersHealth

client_test.go:762–883

(t *testing.T)

}

func TestClientCheckBrokersHealth(t *testing.T) {
	newConnectedBroker := func(t *testing.T) (*Broker, *net.TCPConn, func()) {
		t.Helper()

		listener, err := net.Listen("tcp", "127.0.0.1:0")
		require.NoError(t, err)

		accepted := make(chan *net.TCPConn, 1)
		acceptErr := make(chan error, 1)
		go func() {
			conn, err := listener.Accept()
			if err != nil {
				acceptErr <- err
				return
			}
			accepted <- conn.(*net.TCPConn)
		}()

		conn, err := net.Dial("tcp", listener.Addr().String())
		require.NoError(t, err)

		var serverConn *net.TCPConn
		select {
		case serverConn = <-accepted:
		case err := <-acceptErr:
			require.NoError(t, err)
		case <-time.After(time.Second):
			require.FailNow(t, "timed out waiting for test broker connection")
		}

		broker := NewBroker(listener.Addr().String())
		broker.conn = conn.(*net.TCPConn)
		broker.metricRegistry = metrics.NewRegistry()
		broker.opened.Store(true)

		cleanup := func() {
			_ = listener.Close()
			_ = serverConn.Close()
			_ = broker.Close()
		}

		return broker, serverConn, cleanup
	}

	t.Run("does not wait for brokers that are opening", func(t *testing.T) {
		broker := &Broker{id: 1, addr: "127.0.0.1:9092"}
		broker.lock.Lock()
		defer broker.lock.Unlock()

		client := &client{
			brokers: map[int32]*Broker{broker.ID(): broker},
		}

		done := make(chan struct{})
		go func() {
			client.checkBrokersHealth()
			close(done)
		}()