Use a package level Logger.

By default, it is set to discard everything, but you can assign a new log.Logger instance to sarama.Logger.

broker.go

@@ -5,14 +5,12 @@ import (
 	"io"
 	"net"
 	"sync"
-	"log"
 )
 
 // Broker represents a single Kafka broker connection. All operations on this object are entirely concurrency-safe.
 type Broker struct {
 	id     int32
 	addr   string
-	logger *log.Logger
 
 	correlationID int32
 	conn          net.Conn
@@ -31,8 +29,8 @@ type responsePromise struct {
 
 // NewBroker creates and returns a Broker targetting the given host:port address.
 // This does not attempt to actually connect, you have to call Open() for that.
-func NewBroker(addr string, logger *log.Logger) *Broker {
-	return &Broker{id: -1, addr: addr, logger: logger}
+func NewBroker(addr string) *Broker {
+	return &Broker{id: -1, addr: addr}
 }
 
 // Open tries to connect to the Broker. It takes the broker lock synchronously, then spawns a goroutine which
@@ -44,10 +42,8 @@ func (b *Broker) Open() error {
 
 	if b.conn != nil {
 		b.lock.Unlock()
-		if b.logger != nil {
-			b.logger.Printf("Failed to connect to broker %s\n", b.addr)
-			b.logger.Println(AlreadyConnected)
-		}
+		Logger.Printf("Failed to connect to broker %s\n", b.addr)
+		Logger.Println(AlreadyConnected)
 		return AlreadyConnected
 	}
 
@@ -56,10 +52,8 @@ func (b *Broker) Open() error {
 
 		b.conn, b.connErr = net.Dial("tcp", b.addr)
 		if b.connErr != nil {
-			if b.logger != nil {
-				b.logger.Printf("Failed to connect to broker %s\n", b.addr)
-				b.logger.Println(b.connErr)
-			}
+			Logger.Printf("Failed to connect to broker %s\n", b.addr)
+			Logger.Println(b.connErr)
 			return
 		}
 
@@ -68,10 +62,7 @@ func (b *Broker) Open() error {
 		// permit a few outstanding requests before we block waiting for responses
 		b.responses = make(chan responsePromise, 4)
 
-		if b.logger != nil {
-			b.logger.Printf("Connected to broker %s\n", b.addr)
-		}
-
+		Logger.Printf("Connected to broker %s\n", b.addr)
 		go b.responseReceiver()
 	}()
 
@@ -90,16 +81,14 @@ func (b *Broker) Connected() (bool, error) {
 func (b *Broker) Close() (err error) {
 	b.lock.Lock()
 	defer b.lock.Unlock()
-	if b.logger != nil {
-		defer func() {
-			if err == nil {
-				b.logger.Printf("Closed connection to broker #%d %s\n", b.id, b.addr)
-			} else {
-				b.logger.Printf("Failed to close connection to broker #%d %s.\n", b.id, b.addr)
-				b.logger.Println(err)
-			}
-		}()
-	}
+	defer func() {
+		if err == nil {
+			Logger.Printf("Closed connection to broker #%d %s\n", b.id, b.addr)
+		} else {
+			Logger.Printf("Failed to close connection to broker #%d %s.\n", b.id, b.addr)
+			Logger.Println(err)
+		}
+	}()
 
 	if b.conn == nil {
 		return NotConnected

broker_test.go

@@ -143,7 +143,7 @@ func NewMockBroker(t *testing.T, responses chan []byte) *MockBroker {
 }
 
 func ExampleBroker() error {
-	broker := NewBroker("localhost:9092", nil)
+	broker := NewBroker("localhost:9092")
 	err := broker.Open()
 	if err != nil {
 		return err
@@ -162,8 +162,7 @@ func ExampleBroker() error {
 }
 
 func TestBrokerAccessors(t *testing.T) {
-
-	broker := NewBroker("abc:123", nil)
+	broker := NewBroker("abc:123")
 
 	if broker.ID() != -1 {
 		t.Error("New broker didn't have an ID of -1.")
@@ -184,7 +183,7 @@ func TestSimpleBrokerCommunication(t *testing.T) {
 	mockBroker := NewMockBroker(t, responses)
 	defer mockBroker.Close()
 
-	broker := NewBroker(mockBroker.Addr(), nil)
+	broker := NewBroker(mockBroker.Addr())
 	err := broker.Open()
 	if err != nil {
 		t.Fatal(err)

client.go

@@ -4,14 +4,12 @@ import (
 	"sort"
 	"sync"
 	"time"
-	"log"
 )
 
 // ClientConfig is used to pass multiple configuration options to NewClient.
 type ClientConfig struct {
 	MetadataRetries int           // How many times to retry a metadata request when a partition is in the middle of leader election.
 	WaitForElection time.Duration // How long to wait for leader election to finish between retries.
-	Logger          *log.Logger   // The interface to log broker connection events to.
 }
 
 // Client is a generic Kafka client. It manages connections to one or more Kafka brokers.
@@ -53,7 +51,7 @@ func NewClient(id string, addrs []string, config *ClientConfig) (*Client, error)
 		id:               id,
 		config:           *config,
 		extraBrokerAddrs: addrs,
-		extraBroker:      NewBroker(addrs[0], config.Logger),
+		extraBroker:      NewBroker(addrs[0]),
 		brokers:          make(map[int32]*Broker),
 		leaders:          make(map[string]map[int32]int32),
 	}
@@ -166,7 +164,7 @@ func (client *Client) disconnectBroker(broker *Broker) {
 	if broker == client.extraBroker {
 		client.extraBrokerAddrs = client.extraBrokerAddrs[1:]
 		if len(client.extraBrokerAddrs) > 0 {
-			client.extraBroker = NewBroker(client.extraBrokerAddrs[0], client.config.Logger)
+			client.extraBroker = NewBroker(client.extraBrokerAddrs[0])
 			client.extraBroker.Open()
 		} else {
 			client.extraBroker = nil

sarama.go

@@ -5,3 +5,12 @@ The Request/Response objects and properties are mostly undocumented, as they lin
 protocol fields documented by Kafka at https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol
 */
 package sarama
+
+import (
+  "log"
+  "io/ioutil"
+)
+
+var (
+  Logger = log.New(ioutil.Discard, "[Sarama] ", log.LstdFlags)
+)