Function TestClientMetadataTimeout

client_test.go:932–1018

(t *testing.T)

}

func TestClientMetadataTimeout(t *testing.T) {
	tests := []struct {
		name    string
		timeout time.Duration
	}{
		{
			"timeout=250ms",
			250 * time.Millisecond, // Will cut the first retry pass
		},
		{
			"timeout=500ms",
			500 * time.Millisecond, // Will cut the second retry pass
		},
		{
			"timeout=750ms",
			750 * time.Millisecond, // Will cut the third retry pass
		},
		{
			"timeout=900ms",
			900 * time.Millisecond, // Will stop after the three retries
		},
	}

	for _, singleFlight := range []bool{true, false} {
		for _, tc := range tests {
			t.Run(fmt.Sprintf("%s_singleflight_is_%t", tc.name, singleFlight), func(t *testing.T) {
				// Use a responsive broker to create a working client
				initialSeed := NewMockBroker(t, 0)
				emptyMetadata := new(MetadataResponse)
				emptyMetadata.AddBroker(initialSeed.Addr(), initialSeed.BrokerID())
				initialSeed.Returns(emptyMetadata)

				conf := NewTestConfig()
				// Speed up the metadata request failure because of a read timeout
				conf.Net.ReadTimeout = 100 * time.Millisecond
				// Disable backoff and refresh
				conf.Metadata.Retry.Backoff = 0
				conf.Metadata.RefreshFrequency = 0
				// But configure a "global" timeout
				conf.Metadata.Timeout = tc.timeout
				conf.Metadata.SingleFlight = singleFlight
				c, err := NewClient([]string{initialSeed.Addr()}, conf)
				if err != nil {
					t.Fatal(err)
				}
				initialSeed.Close()

				client := c.(*client)

				// Start seed brokers that do not reply to anything and therefore a read
				// on the TCP connection will timeout to simulate unresponsive brokers
				seed1 := NewMockBroker(t, 1)
				defer seed1.Close()
				seed2 := NewMockBroker(t, 2)
				defer seed2.Close()

				// Overwrite the seed brokers with a fixed ordering to make this test deterministic
				safeClose(t, client.seedBrokers[0])