Register new brokers from consumer metadata responses.

IBM · Mar 21, 2015 · b5466ce · b5466ce
1 parent c0cd9ce
commit b5466ce
Showing 1 changed file with 10 additions and 2 deletions.
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]
+		broker, ok := client.brokers[response.CoordinatorID]
+		if !ok {
+			broker = &Broker{
+				id:   response.CoordinatorID,
+				addr: fmt.Sprintf("%s:%d", response.CoordinatorHost, response.CoordinatorPort),
+			}
+
+			client.brokers[response.CoordinatorID] = broker
+		}
 		_ = broker.Open(client.conf)
 		return broker, nil