Function TestClusterAdminListTopicsRetriesOnTransientConnectionError

admin_test.go:209–276

(t *testing.T)

}

func TestClusterAdminListTopicsRetriesOnTransientConnectionError(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)
	defer seedBroker.Close()

	metadataResponse := NewMockMetadataResponse(t).
		SetController(seedBroker.BrokerID()).
		SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
		SetLeader("my_topic", 0, seedBroker.BrokerID())

	var stateMu sync.Mutex
	var injectTimeout, injected bool
	var metadataAttempts int

	seedBroker.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"MetadataRequest": func(req *request) encoderWithHeader {
			stateMu.Lock()
			metadataAttempts++
			shouldInject := injectTimeout && !injected
			if shouldInject {
				injected = true
			}
			stateMu.Unlock()
			if shouldInject {
				return nil
			}
			return metadataResponse.For(req.body)
		},
		"DescribeConfigsRequest": func(req *request) encoderWithHeader {
			return NewMockDescribeConfigsResponse(t).For(req.body)
		},
	})

	config := NewTestConfig()
	config.Version = V1_1_0_0
	config.Net.ReadTimeout = 100 * time.Millisecond
	config.Admin.Retry.Max = 3
	config.Admin.Retry.Backoff = 10 * time.Millisecond

	admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
	if err != nil {
		t.Fatal(err)
	}
	defer safeClose(t, admin)

	stateMu.Lock()
	injectTimeout = true
	metadataAttemptsBeforeList := metadataAttempts
	stateMu.Unlock()

	entries, err := admin.ListTopics()
	if err != nil {
		t.Fatal(err)
	}

	stateMu.Lock()
	gotInjected := injected
	gotMetadataAttempts := metadataAttempts
	stateMu.Unlock()