Correctly send client ID

broker.go

@@ -148,11 +148,11 @@ func (b *broker) rcvResponseLoop() {
 	}
 }
 
-func (b *broker) sendRequest(api API, body encoder) (chan []byte, error) {
+func (b *broker) sendRequest(clientID *string, api API, body encoder) (chan []byte, error) {
 	var prepEnc prepEncoder
 	var realEnc realEncoder
 
-	req := request{api, b.correlation_id, nil, body}
+	req := request{api, b.correlation_id, clientID, body}
 
 	req.encode(&prepEnc)
 	if prepEnc.err {

brokerManager.go

@@ -11,14 +11,17 @@ type brokerKey struct {
 }
 
 type brokerManager struct {
+	client        *Client
 	defaultBroker *broker
 	leaders       map[brokerKey]*broker
 	leadersLock   sync.RWMutex
 }
 
-func newBrokerManager(host string, port int32) (bm *brokerManager, err error) {
+func newBrokerManager(client *Client, host string, port int32) (bm *brokerManager, err error) {
 	bm = new(brokerManager)
 
+	bm.client = client
+
 	bm.defaultBroker, err = newBroker(host, port)
 	if err != nil {
 		return nil, err
@@ -58,7 +61,7 @@ func (bm *brokerManager) refreshTopics(topics []*string) error {
 		return errors.New("kafka: lost all broker connections")
 	}
 
-	responseChan, err := b.sendRequest(REQUEST_METADATA, &metadataRequest{topics})
+	responseChan, err := b.sendRequest(bm.client.id, REQUEST_METADATA, &metadataRequest{topics})
 	if err != nil {
 		// TODO
 	}

client.go

@@ -8,7 +8,7 @@ type Client struct {
 func NewClient(id *string, host string, port int32) (client *Client, err error) {
 	client = new(Client)
 	client.id = id
-	client.brokers, err = newBrokerManager(host, port)
+	client.brokers, err = newBrokerManager(client, host, port)
 	if err != nil {
 		return nil, err
 	}