Register new brokers from consumer metadata responses.

IBM · Mar 21, 2015 · d441c77 · d441c77
1 parent c0cd9ce
commit d441c77
Showing 1 changed file with 9 additions and 1 deletion.
diff --git a/client.go b/client.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"fmt"
 	"sort"
 	"sync"
 	"time"
@@ -382,8 +383,15 @@ func (client *client) getOffsetLeader(consumerGroup string, attemptsRemaining in
 
 	switch response.Err {
 	case ErrNoError:
-		// TODO: register new brokers
 		broker := client.brokers[response.CoordinatorID]
+		if broker == nil {
+			client.brokers[response.CoordinatorID] = &Broker{
+				id:   response.CoordinatorID,
+				addr: fmt.Sprintf("%s:%d", response.CoordinatorHost, response.CoordinatorPort),
+			}
+
+			broker = client.brokers[response.CoordinatorID]
+		}
 		_ = broker.Open(client.conf)
 		return broker, nil