client: merge Request and Admin for simplicity
@@ -292,23 +292,16 @@ func (c *Client) updateBrokers(brokers []kmsg.MetadataResponseBrokers) {
c.anyBroker = newAnyBroker
}
+// Request issues a request to Kafka, waiting for and returning the response or
+// an error.
+//
+// If the request is an admin request, this will issue it to the Kafka
+// controller. If the controller ID is unknown, this will attempt to fetch it.
+// If the fetch errors, this will return an unknown controller error.
func (c *Client) Request(req kmsg.Request) (kmsg.Response, error) {
- broker := c.broker()
+ var broker *broker
- var resp kmsg.Response
- var err error
- broker.wait(req, func(kresp kmsg.Response, kerr error) {
- resp, err = kresp, kerr
- })
- return resp, err
-}
-
-// Admin issues an admin request to the controller broker, waiting for and
-// returning the Kafka response or an error.
-//
-// If the controller ID is unknown, this will attempt to fetch it. If the
-// fetch errors, this will return an unknown controller error.
-func (c *Client) Admin(req kmsg.AdminRequest) (kmsg.Response, error) {
+ if _, admin := req.(kmsg.AdminRequest); admin {
retries := 0
start:
if c.controllerID < 0 {
@@ -332,10 +325,15 @@ start:
if !exists {
return nil, errUnknownController
}
+ broker = controller
+
+ } else {
+ broker = c.broker()
+ }
var resp kmsg.Response
var err error
- controller.wait(req, func(kresp kmsg.Response, kerr error) {
+ broker.wait(req, func(kresp kmsg.Response, kerr error) {
resp, err = kresp, kerr
})
return resp, err
