Function singleBrokerAdmin

admin_test.go:1457–1478

singleBrokerAdmin wires a single mock broker that names itself controller, installs the given per-request handlers, and returns a ready admin client.

(t *testing.T, version KafkaVersion, handlers map[string]requestHandlerFunc)

// singleBrokerAdmin wires a single mock broker that names itself controller,
// installs the given per-request handlers, and returns a ready admin client.
func singleBrokerAdmin(t *testing.T, version KafkaVersion, handlers map[string]requestHandlerFunc) ClusterAdmin {
	b := NewMockBroker(t, 1)
	t.Cleanup(func() { b.Close() })

	hm := map[string]requestHandlerFunc{
		"MetadataRequest": func(req *request) encoderWithHeader {
			return NewMockMetadataResponse(t).
				SetController(b.BrokerID()).
				SetBroker(b.Addr(), b.BrokerID()).For(req.body)
		},
	}
	maps.Copy(hm, handlers)
	b.SetHandlerFuncByMap(hm)

	config := NewTestConfig()
	config.Version = version
	admin, err := NewClusterAdmin([]string{b.Addr()}, config)
	require.NoError(t, err)
	t.Cleanup(func() { _ = admin.Close() })

	return admin
}

// staleControllerAdmin wires two mock brokers into a controller-failover
// scenario and returns a ready admin client. Broker 1 answers reqType with