Function staleControllerAdmin

admin_test.go:1485–1522

staleControllerAdmin wires two mock brokers into a controller-failover scenario and returns a ready admin client. Broker 1 answers reqType with errResp and flips fresh metadata to name broker 2 as controller, modeling a stale cached controller; broker 2 answers with okResp. retriedOnNewController re

(t *testing.T, version KafkaVersion, reqType string, errResp, okResp requestHandlerFunc)

// stale cached controller; broker 2 answers with okResp. retriedOnNewController
// reports whether broker 2 received the request.
func staleControllerAdmin(t *testing.T, version KafkaVersion, reqType string, errResp, okResp requestHandlerFunc) (admin ClusterAdmin, retriedOnNewController func() bool) {
	b1 := NewMockBroker(t, 1)
	b2 := NewMockBroker(t, 2)
	t.Cleanup(func() { b1.Close(); b2.Close() })

	var controllerID atomic.Int32
	controllerID.Store(b1.BrokerID())
	meta := func(req *request) encoderWithHeader {
		return NewMockMetadataResponse(t).SetController(controllerID.Load()).
			SetBroker(b1.Addr(), b1.BrokerID()).
			SetBroker(b2.Addr(), b2.BrokerID()).For(req.body)
	}

	var newControllerReceivedRequest atomic.Bool
	b1.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"MetadataRequest": meta,
		reqType: func(req *request) encoderWithHeader {
			controllerID.Store(b2.BrokerID()) // flip so the post-error refresh elects broker 2
			return errResp(req)
		},
	})
	b2.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"MetadataRequest": meta,
		reqType: func(req *request) encoderWithHeader {
			newControllerReceivedRequest.Store(true)
			return okResp(req)
		},
	})

	config := NewTestConfig()
	config.Version = version
	config.Admin.Retry.Backoff = time.Millisecond
	admin, err := NewClusterAdmin([]string{b1.Addr()}, config)
	require.NoError(t, err)
	t.Cleanup(func() { _ = admin.Close() })

	return admin, newControllerReceivedRequest.Load
}

func TestClusterAdminListAcls(t *testing.T) {
	resourceName := "my_topic"