Merge pull request #324 from Shopify/mock_package

Add basic MockProducer

mocks/mocks.go

@@ -0,0 +1,33 @@
+/*
+Package mocks provides mocks that can be used for testing applications
+that use Sarama. The mock types provided by this package implement the
+interfaces Sarama exports, so you can use them for dependency injection
+in your tests.
+
+All mock instances require you to set expectations on them before you
+can use them. It will determine how the mock will behave. If an
+expectation is not met, it will make your test fail.
+
+NOTE: this package currently does not fall under the API stability
+guarantee of Sarama as it is still considered experimental.
+*/
+package mocks
+
+import (
+	"errors"
+)
+
+// A simple interface that includes the testing.T methods we use to report
+// expectation violations when using the mock objects.
+type ErrorReporter interface {
+	Errorf(string, ...interface{})
+}
+
+var (
+	errProduceSuccess    error = nil
+	errOutOfExpectations       = errors.New("No more expectations set on mock producer")
+)
+
+type producerExpectation struct {
+	Result error
+}

mocks/producer.go

@@ -0,0 +1,137 @@
+package mocks
+
+import (
+	"sync"
+
+	"github.com/Shopify/sarama"
+)
+
+// Producer implements sarama's Producer interface for testing purposes.
+// Before you can send messages to it's Input channel, you have to set expectations
+// so it knows how to handle the input. This way you can easily test success and
+// failure scenarios.
+type Producer struct {
+	l            sync.Mutex
+	t            ErrorReporter
+	expectations []*producerExpectation
+	closed       chan struct{}
+	input        chan *sarama.ProducerMessage
+	successes    chan *sarama.ProducerMessage
+	errors       chan *sarama.ProducerError
+}
+
+// NewProducer instantiates a new Producer mock. The t argument should
+// be the *testing.T instance of your test method. An error will be written to it if
+// an expectation is violated. The config argument is used to determine whether it
+// should ack successes on the Successes channel.
+func NewProducer(t ErrorReporter, config *sarama.Config) *Producer {
+	if config == nil {
+		config = sarama.NewConfig()
+	}
+	mp := &Producer{
+		t:            t,
+		closed:       make(chan struct{}, 0),
+		expectations: make([]*producerExpectation, 0),
+		input:        make(chan *sarama.ProducerMessage, config.ChannelBufferSize),
+		successes:    make(chan *sarama.ProducerMessage, config.ChannelBufferSize),
+		errors:       make(chan *sarama.ProducerError, config.ChannelBufferSize),
+	}
+
+	go func() {
+		defer func() {
+			close(mp.successes)
+			close(mp.errors)
+		}()
+
+		for msg := range mp.input {
+			mp.l.Lock()
+			if mp.expectations == nil || len(mp.expectations) == 0 {
+				mp.expectations = nil
+				mp.t.Errorf("No more expectation set on this mock producer to handle the input message.")
+			} else {
+				expectation := mp.expectations[0]
+				mp.expectations = mp.expectations[1:]
+				if expectation.Result == errProduceSuccess {
+					if config.Producer.AckSuccesses {
+						mp.successes <- msg
+					}
+				} else {
+					mp.errors <- &sarama.ProducerError{Err: expectation.Result, Msg: msg}
+				}
+			}
+			mp.l.Unlock()
+		}
+
+		mp.l.Lock()
+		if len(mp.expectations) > 0 {
+			mp.t.Errorf("Expected to exhaust all expectations, but %d are left.", len(mp.expectations))
+		}
+		mp.l.Unlock()
+
+		close(mp.closed)
+	}()
+
+	return mp
+}
+
+////////////////////////////////////////////////
+// Implement Producer interface
+////////////////////////////////////////////////
+
+// AsyncClose corresponds with the AsyncClose method of sarama's Producer implementation.
+// By closing a mock producer, you also tell it that no more input will be provided, so it will
+// write an error to the test state if there's any remaining expectations.
+func (mp *Producer) AsyncClose() {
+	close(mp.input)
+}
+
+// Close corresponds with the Close method of sarama's Producer implementation.
+// By closing a mock producer, you also tell it that no more input will be provided, so it will
+// write an error to the test state if there's any remaining expectations.
+func (mp *Producer) Close() error {
+	mp.AsyncClose()
+	<-mp.closed
+	return nil
+}
+
+// Input corresponds with the Input method of sarama's Producer implementation.
+// You have to set expectations on the mock producer before writing messages to the Input
+// channel, so it knows how to handle them. If there is no more remaining expectations and
+// a messages is written to the Input channel, the mock producer will write an error to the test
+// state object.
+func (mp *Producer) Input() chan<- *sarama.ProducerMessage {
+	return mp.input
+}
+
+// Successes corresponds with the Successes method of sarama's Producer implementation.
+func (mp *Producer) Successes() <-chan *sarama.ProducerMessage {
+	return mp.successes
+}
+
+// Errors corresponds with the Errors method of sarama's Producer implementation.
+func (mp *Producer) Errors() <-chan *sarama.ProducerError {
+	return mp.errors
+}
+
+////////////////////////////////////////////////
+// Setting expectations
+////////////////////////////////////////////////
+
+// ExpectInputAndSucceed sets an expectation on the mock producer that a message will be provided
+// on the input channel. The mock producer will handle the message as if it is produced successfully,
+// i.e. it will make it available on the Successes channel if the Producer.AckSuccesses setting
+// is set to true.
+func (mp *Producer) ExpectInputAndSucceed() {
+	mp.l.Lock()
+	defer mp.l.Unlock()
+	mp.expectations = append(mp.expectations, &producerExpectation{Result: errProduceSuccess})
+}
+
+// ExpectInputAndFail sets an expectation on the mock producer that a message will be provided
+// on the input channel. The mock producer will handle the message as if it failed to produce
+// successfully. This means it will make a ProducerError available on the Errors channel.
+func (mp *Producer) ExpectInputAndFail(err error) {
+	mp.l.Lock()
+	defer mp.l.Unlock()
+	mp.expectations = append(mp.expectations, &producerExpectation{Result: err})
+}

mocks/producer_test.go

@@ -0,0 +1,88 @@
+package mocks
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/Shopify/sarama"
+)
+
+type testReporterMock struct {
+	errors []string
+}
+
+func newTestReporterMock() *testReporterMock {
+	return &testReporterMock{errors: make([]string, 0)}
+}
+
+func (trm *testReporterMock) Errorf(format string, args ...interface{}) {
+	trm.errors = append(trm.errors, fmt.Sprintf(format, args...))
+}
+
+func TestMockProducerImplementsProducerInterface(t *testing.T) {
+	var mp interface{} = &Producer{}
+	if _, ok := mp.(sarama.Producer); !ok {
+		t.Error("The mock producer should implement the sarama.Producer interface.")
+	}
+}
+
+func TestProducerReturnsExpectationsToChannels(t *testing.T) {
+	config := sarama.NewConfig()
+	config.Producer.AckSuccesses = true
+	mp := NewProducer(t, config)
+
+	mp.ExpectInputAndSucceed()
+	mp.ExpectInputAndSucceed()
+	mp.ExpectInputAndFail(sarama.ErrOutOfBrokers)
+
+	mp.Input() <- &sarama.ProducerMessage{Topic: "test 1"}
+	mp.Input() <- &sarama.ProducerMessage{Topic: "test 2"}
+	mp.Input() <- &sarama.ProducerMessage{Topic: "test 3"}
+
+	msg1 := <-mp.Successes()
+	msg2 := <-mp.Successes()
+	err1 := <-mp.Errors()
+
+	if msg1.Topic != "test 1" {
+		t.Error("Expected message 1 to be returned first")
+	}
+
+	if msg2.Topic != "test 2" {
+		t.Error("Expected message 2 to be returned second")
+	}
+
+	if err1.Msg.Topic != "test 3" || err1.Err != sarama.ErrOutOfBrokers {
+		t.Error("Expected message 3 to be returned as error")
+	}
+
+	mp.Close()
+}
+
+func TestProducerWithTooFewExpectations(t *testing.T) {
+	trm := newTestReporterMock()
+	mp := NewProducer(trm, nil)
+	mp.ExpectInputAndSucceed()
+
+	mp.Input() <- &sarama.ProducerMessage{Topic: "test"}
+	mp.Input() <- &sarama.ProducerMessage{Topic: "test"}
+
+	mp.Close()
+
+	if len(trm.errors) != 1 {
+		t.Error("Expected to report an error")
+	}
+}
+
+func TestProducerWithTooManyExpectations(t *testing.T) {
+	trm := newTestReporterMock()
+	mp := NewProducer(trm, nil)
+	mp.ExpectInputAndSucceed()
+	mp.ExpectInputAndFail(sarama.ErrOutOfBrokers)
+
+	mp.Input() <- &sarama.ProducerMessage{Topic: "test"}
+	mp.Close()
+
+	if len(trm.errors) != 1 {
+		t.Error("Expected to report an error")
+	}
+}

mocks/sync_producer.go

@@ -0,0 +1,92 @@
+package mocks
+
+import (
+	"github.com/Shopify/sarama"
+	"sync"
+)
+
+// SyncProducer implements sarama's SyncProducer interface for testing purposes.
+// Before you can use it, you have to set expectations on the mock SyncProducer
+// to tell it how to handle calls to SendMessage, so you can easily test success
+// and failure scenarios.
+type SyncProducer struct {
+	l            sync.Mutex
+	t            ErrorReporter
+	expectations []*producerExpectation
+	lastOffset   int64
+}
+
+// NewSyncProducer instantiates a new SyncProducer mock. The t argument should
+// be the *testing.T instance of your test method. An error will be written to it if
+// an expectation is violated. The config argument is currently unused, but is
+// maintained to be compatible with the async Producer.
+func NewSyncProducer(t ErrorReporter, config *sarama.Config) *SyncProducer {
+	return &SyncProducer{
+		t:            t,
+		expectations: make([]*producerExpectation, 0),
+	}
+}
+
+////////////////////////////////////////////////
+// Implement SyncProducer interface
+////////////////////////////////////////////////
+
+// SendMessage corresponds with the SendMessage method of sarama's SyncProducer implementation.
+// You have to set expectations on the mock producer before calling SendMessage, so it knows
+// how to handle them. If there is no more remaining expectations when SendMessage is called,
+// the mock producer will write an error to the test state object.
+func (sp *SyncProducer) SendMessage(topic string, key, value sarama.Encoder) (partition int32, offset int64, err error) {
+	sp.l.Lock()
+	defer sp.l.Unlock()
+
+	if len(sp.expectations) > 0 {
+		expectation := sp.expectations[0]
+		sp.expectations = sp.expectations[1:]
+
+		if expectation.Result == errProduceSuccess {
+			sp.lastOffset++
+			return 0, sp.lastOffset, nil
+		} else {
+			return -1, -1, expectation.Result
+		}
+	} else {
+		sp.t.Errorf("No more expectation set on this mock producer to handle the input message.")
+		return -1, -1, errOutOfExpectations
+	}
+}
+
+// Close corresponds with the Close method of sarama's SyncProducer implementation.
+// By closing a mock syncproducer, you also tell it that no more SendMessage calls will follow,
+// so it will write an error to the test state if there's any remaining expectations.
+func (sp *SyncProducer) Close() error {
+	sp.l.Lock()
+	defer sp.l.Unlock()
+
+	if len(sp.expectations) > 0 {
+		sp.t.Errorf("Expected to exhaust all expectations, but %d are left.", len(sp.expectations))
+	}
+
+	return nil
+}
+
+////////////////////////////////////////////////
+// Setting expectations
+////////////////////////////////////////////////
+
+// ExpectSendMessageAndSucceed sets an expectation on the mock producer that SendMessage will be
+// called. The mock producer will handle the message as if it produced successfully, i.e. by
+// returning a valid partition, and offset, and a nil error.
+func (sp *SyncProducer) ExpectSendMessageAndSucceed() {
+	sp.l.Lock()
+	defer sp.l.Unlock()
+	sp.expectations = append(sp.expectations, &producerExpectation{Result: errProduceSuccess})
+}
+
+// ExpectSendMessageAndFail sets an expectation on the mock producer that SendMessage will be
+// called. The mock producer will handle the message as if it failed to produce
+// successfully, i.e. by returning the provided error.
+func (sp *SyncProducer) ExpectSendMessageAndFail(err error) {
+	sp.l.Lock()
+	defer sp.l.Unlock()
+	sp.expectations = append(sp.expectations, &producerExpectation{Result: err})
+}

mocks/sync_producer_test.go

@@ -0,0 +1,83 @@
+package mocks
+
+import (
+	"testing"
+
+	"github.com/Shopify/sarama"
+)
+
+func TestMockSyncProducerImplementsSyncProducerInterface(t *testing.T) {
+	var mp interface{} = &SyncProducer{}
+	if _, ok := mp.(sarama.SyncProducer); !ok {
+		t.Error("The mock async producer should implement the sarama.SyncProducer interface.")
+	}
+}
+
+func TestSyncProducerReturnsExpectationsToSendMessage(t *testing.T) {
+	sp := NewSyncProducer(t, nil)
+	defer sp.Close()
+
+	sp.ExpectSendMessageAndSucceed()
+	sp.ExpectSendMessageAndSucceed()
+	sp.ExpectSendMessageAndFail(sarama.ErrOutOfBrokers)
+
+	var (
+		offset int64
+		err    error
+	)
+
+	_, offset, err = sp.SendMessage("test", nil, sarama.StringEncoder("test"))
+	if err != nil {
+		t.Errorf("The first message should have been produced successfully, but got %s", err)
+	}
+	if offset != 1 {
+		t.Errorf("The first message should have been assigned offset 1, but got %d", offset)
+	}
+
+	_, offset, err = sp.SendMessage("test", nil, sarama.StringEncoder("test"))
+	if err != nil {
+		t.Errorf("The second message should have been produced successfully, but got %s", err)
+	}
+	if offset != 2 {
+		t.Errorf("The second message should have been assigned offset 2, but got %d", offset)
+	}
+
+	_, offset, err = sp.SendMessage("test", nil, sarama.StringEncoder("test"))
+	if err != sarama.ErrOutOfBrokers {
+		t.Errorf("The third message should not have been produced successfully")
+	}
+
+	sp.Close()
+}
+
+func TestSyncProducerWithTooManyExpectations(t *testing.T) {
+	trm := newTestReporterMock()
+
+	sp := NewSyncProducer(trm, nil)
+	sp.ExpectSendMessageAndSucceed()
+	sp.ExpectSendMessageAndFail(sarama.ErrOutOfBrokers)
+
+	sp.SendMessage("test", nil, sarama.StringEncoder("test"))
+
+	sp.Close()
+
+	if len(trm.errors) != 1 {
+		t.Error("Expected to report an error")
+	}
+}
+
+func TestSyncProducerWithTooFewExpectations(t *testing.T) {
+	trm := newTestReporterMock()
+
+	sp := NewSyncProducer(trm, nil)
+	sp.ExpectSendMessageAndSucceed()
+
+	sp.SendMessage("test", nil, sarama.StringEncoder("test"))
+	sp.SendMessage("test", nil, sarama.StringEncoder("test"))
+
+	sp.Close()
+
+	if len(trm.errors) != 1 {
+		t.Error("Expected to report an error")
+	}
+}

producer.go

@@ -19,7 +19,34 @@ func forceFlushThreshold() int {
 // leaks: it will not be garbage-collected automatically when it passes out of
 // scope (this is in addition to calling Close on the underlying client, which
 // is still necessary).
-type Producer struct {
+type Producer interface {
+
+	// AsyncClose triggers a shutdown of the producer, flushing any messages it may have
+	// buffered. The shutdown has completed when both the Errors and Successes channels
+	// have been closed. When calling AsyncClose, you *must* continue to read from those
+	// channels in order to drain the results of any messages in flight.
+	AsyncClose()
+
+	// Close shuts down the producer and flushes any messages it may have buffered.
+	// You must call this function before a producer object passes out of scope, as
+	// it may otherwise leak memory. You must call this before calling Close on the
+	// underlying client.
+	Close() error
+
+	// Input is the input channel for the user to write messages to that they wish to send.
+	Input() chan<- *ProducerMessage
+
+	// Successes is the success output channel back to the user when AckSuccesses is confured.
+	// If AckSuccesses is true, you MUST read from this channel or the Producer will deadlock.
+	// It is suggested that you send and read messages together in a single select statement.
+	Successes() <-chan *ProducerMessage
+
+	// Errors is the error output channel back to the user. You MUST read from this channel or the Producer will deadlock.
+	// It is suggested that you send messages and read errors together in a single select statement.
+	Errors() <-chan *ProducerError
+}
+
+type producer struct {
 	client    *Client
 	conf      *Config
 	ownClient bool
@@ -32,7 +59,7 @@ type Producer struct {
 }
 
 // NewProducer creates a new Producer using the given broker addresses and configuration.
-func NewProducer(addrs []string, conf *Config) (*Producer, error) {
+func NewProducer(addrs []string, conf *Config) (Producer, error) {
 	client, err := NewClient(addrs, conf)
 	if err != nil {
 		return nil, err
@@ -42,18 +69,18 @@ func NewProducer(addrs []string, conf *Config) (*Producer, error) {
 	if err != nil {
 		return nil, err
 	}
-	p.ownClient = true
+	p.(*producer).ownClient = true
 	return p, nil
 }
 
 // NewProducerFromClient creates a new Producer using the given client.
-func NewProducerFromClient(client *Client) (*Producer, error) {
+func NewProducerFromClient(client *Client) (Producer, error) {
 	// Check that we are not dealing with a closed Client before processing any other arguments
 	if client.Closed() {
 		return nil, ErrClosedClient
 	}
 
-	p := &Producer{
+	p := &producer{
 		client:    client,
 		conf:      client.conf,
 		errors:    make(chan *ProducerError),
@@ -136,29 +163,19 @@ func (pe ProducerErrors) Error() string {
 	return fmt.Sprintf("kafka: Failed to deliver %d messages.", len(pe))
 }
 
-// Errors is the error output channel back to the user. You MUST read from this channel or the Producer will deadlock.
-// It is suggested that you send messages and read errors together in a single select statement.
-func (p *Producer) Errors() <-chan *ProducerError {
+func (p *producer) Errors() <-chan *ProducerError {
 	return p.errors
 }
 
-// Successes is the success output channel back to the user when AckSuccesses is confured.
-// If AckSuccesses is true, you MUST read from this channel or the Producer will deadlock.
-// It is suggested that you send and read messages together in a single select statement.
-func (p *Producer) Successes() <-chan *ProducerMessage {
+func (p *producer) Successes() <-chan *ProducerMessage {
 	return p.successes
 }
 
-// Input is the input channel for the user to write messages to that they wish to send.
-func (p *Producer) Input() chan<- *ProducerMessage {
+func (p *producer) Input() chan<- *ProducerMessage {
 	return p.input
 }
 
-// Close shuts down the producer and flushes any messages it may have buffered.
-// You must call this function before a producer object passes out of scope, as
-// it may otherwise leak memory. You must call this before calling Close on the
-// underlying client.
-func (p *Producer) Close() error {
+func (p *producer) Close() error {
 	p.AsyncClose()
 
 	if p.conf.Producer.AckSuccesses {
@@ -179,11 +196,7 @@ func (p *Producer) Close() error {
 	return nil
 }
 
-// AsyncClose triggers a shutdown of the producer, flushing any messages it may have
-// buffered. The shutdown has completed when both the Errors and Successes channels
-// have been closed. When calling AsyncClose, you *must* continue to read from those
-// channels in order to drain the results of any messages in flight.
-func (p *Producer) AsyncClose() {
+func (p *producer) AsyncClose() {
 	go withRecover(func() {
 		p.input <- &ProducerMessage{flags: shutdown}
 	})
@@ -198,7 +211,7 @@ func (p *Producer) AsyncClose() {
 
 // singleton
 // dispatches messages by topic
-func (p *Producer) topicDispatcher() {
+func (p *producer) topicDispatcher() {
 	handlers := make(map[string]chan *ProducerMessage)
 
 	for msg := range p.input {
@@ -254,7 +267,7 @@ func (p *Producer) topicDispatcher() {
 
 // one per topic
 // partitions messages, then dispatches them by partition
-func (p *Producer) partitionDispatcher(topic string, input chan *ProducerMessage) {
+func (p *producer) partitionDispatcher(topic string, input chan *ProducerMessage) {
 	handlers := make(map[int32]chan *ProducerMessage)
 	partitioner := p.conf.Producer.Partitioner()
 
@@ -290,7 +303,7 @@ func (p *Producer) partitionDispatcher(topic string, input chan *ProducerMessage
 // one per partition per topic
 // dispatches messages to the appropriate broker
 // also responsible for maintaining message order during retries
-func (p *Producer) leaderDispatcher(topic string, partition int32, input chan *ProducerMessage) {
+func (p *producer) leaderDispatcher(topic string, partition int32, input chan *ProducerMessage) {
 	var leader *Broker
 	var output chan *ProducerMessage
 
@@ -407,7 +420,7 @@ func (p *Producer) leaderDispatcher(topic string, partition int32, input chan *P
 // one per broker
 // groups messages together into appropriately-sized batches for sending to the broker
 // based on https://godoc.org/github.com/eapache/channels#BatchingChannel
-func (p *Producer) messageAggregator(broker *Broker, input chan *ProducerMessage) {
+func (p *producer) messageAggregator(broker *Broker, input chan *ProducerMessage) {
 	var ticker *time.Ticker
 	var timer <-chan time.Time
 	if p.conf.Producer.Flush.Frequency > 0 {
@@ -467,7 +480,7 @@ shutdown:
 
 // one per broker
 // takes a batch at a time from the messageAggregator and sends to the broker
-func (p *Producer) flusher(broker *Broker, input chan []*ProducerMessage) {
+func (p *producer) flusher(broker *Broker, input chan []*ProducerMessage) {
 	var closing error
 	currentRetries := make(map[string]map[int32]error)
 	Logger.Printf("producer/flusher/%d starting up\n", broker.ID())
@@ -573,7 +586,7 @@ func (p *Producer) flusher(broker *Broker, input chan []*ProducerMessage) {
 // singleton
 // effectively a "bridge" between the flushers and the topicDispatcher in order to avoid deadlock
 // based on https://godoc.org/github.com/eapache/channels#InfiniteChannel
-func (p *Producer) retryHandler() {
+func (p *producer) retryHandler() {
 	var buf []*ProducerMessage
 	var msg *ProducerMessage
 	refs := 0
@@ -620,7 +633,7 @@ func (p *Producer) retryHandler() {
 
 // utility functions
 
-func (p *Producer) assignPartition(partitioner Partitioner, msg *ProducerMessage) error {
+func (p *producer) assignPartition(partitioner Partitioner, msg *ProducerMessage) error {
 	var partitions []int32
 	var err error
 
@@ -653,7 +666,7 @@ func (p *Producer) assignPartition(partitioner Partitioner, msg *ProducerMessage
 	return nil
 }
 
-func (p *Producer) buildRequest(batch map[string]map[int32][]*ProducerMessage) *ProduceRequest {
+func (p *producer) buildRequest(batch map[string]map[int32][]*ProducerMessage) *ProduceRequest {
 
 	req := &ProduceRequest{RequiredAcks: p.conf.Producer.RequiredAcks, Timeout: int32(p.conf.Producer.Timeout / time.Millisecond)}
 	empty := true
@@ -715,13 +728,13 @@ func (p *Producer) buildRequest(batch map[string]map[int32][]*ProducerMessage) *
 	return req
 }
 
-func (p *Producer) returnError(msg *ProducerMessage, err error) {
+func (p *producer) returnError(msg *ProducerMessage, err error) {
 	msg.flags = 0
 	msg.retries = 0
 	p.errors <- &ProducerError{Msg: msg, Err: err}
 }
 
-func (p *Producer) returnErrors(batch []*ProducerMessage, err error) {
+func (p *producer) returnErrors(batch []*ProducerMessage, err error) {
 	for _, msg := range batch {
 		if msg != nil {
 			p.returnError(msg, err)
@@ -729,7 +742,7 @@ func (p *Producer) returnErrors(batch []*ProducerMessage, err error) {
 	}
 }
 
-func (p *Producer) returnSuccesses(batch []*ProducerMessage) {
+func (p *producer) returnSuccesses(batch []*ProducerMessage) {
 	for _, msg := range batch {
 		if msg != nil {
 			msg.flags = 0
@@ -738,7 +751,7 @@ func (p *Producer) returnSuccesses(batch []*ProducerMessage) {
 	}
 }
 
-func (p *Producer) retryMessages(batch []*ProducerMessage, err error) {
+func (p *producer) retryMessages(batch []*ProducerMessage, err error) {
 	for _, msg := range batch {
 		if msg == nil {
 			continue
@@ -757,7 +770,7 @@ type brokerProducer struct {
 	refs  int
 }
 
-func (p *Producer) getBrokerProducer(broker *Broker) chan *ProducerMessage {
+func (p *producer) getBrokerProducer(broker *Broker) chan *ProducerMessage {
 	p.brokerLock.Lock()
 	defer p.brokerLock.Unlock()
 
@@ -778,7 +791,7 @@ func (p *Producer) getBrokerProducer(broker *Broker) chan *ProducerMessage {
 	return producer.input
 }
 
-func (p *Producer) unrefBrokerProducer(broker *Broker) {
+func (p *producer) unrefBrokerProducer(broker *Broker) {
 	p.brokerLock.Lock()
 	defer p.brokerLock.Unlock()
 

producer_test.go

@@ -8,7 +8,7 @@ import (
 
 const TestMessage = "ABC THE MESSAGE"
 
-func closeProducer(t *testing.T, p *Producer) {
+func closeProducer(t *testing.T, p Producer) {
 	var wg sync.WaitGroup
 	p.AsyncClose()
 

sync_producer.go

@@ -5,32 +5,43 @@ import "sync"
 // SyncProducer publishes Kafka messages. It routes messages to the correct broker, refreshing metadata as appropriate,
 // and parses responses for errors. You must call Close() on a producer to avoid leaks, it may not be garbage-collected automatically when
 // it passes out of scope (this is in addition to calling Close on the underlying client, which is still necessary).
-type SyncProducer struct {
-	producer *Producer
+type SyncProducer interface {
+	// SendMessage produces a message to the given topic with the given key and value. To send strings as either key or value, see the StringEncoder type.
+	// It returns the partition and offset of the successfully-produced message, or the error (if any).
+	SendMessage(topic string, key, value Encoder) (partition int32, offset int64, err error)
+
+	// Close shuts down the producer and flushes any messages it may have buffered. You must call this function before
+	// a producer object passes out of scope, as it may otherwise leak memory. You must call this before calling Close
+	// on the underlying client.
+	Close() error
+}
+
+type syncProducer struct {
+	producer *producer
 	wg       sync.WaitGroup
 }
 
 // NewSyncProducer creates a new SyncProducer using the given broker addresses and configuration.
-func NewSyncProducer(addrs []string, config *Config) (*SyncProducer, error) {
+func NewSyncProducer(addrs []string, config *Config) (SyncProducer, error) {
 	p, err := NewProducer(addrs, config)
 	if err != nil {
 		return nil, err
 	}
-	return newSyncProducerFromProducer(p), nil
+	return newSyncProducerFromProducer(p.(*producer)), nil
 }
 
 // NewSyncProducerFromClient creates a new SyncProducer using the given client.
-func NewSyncProducerFromClient(client *Client) (*SyncProducer, error) {
+func NewSyncProducerFromClient(client *Client) (SyncProducer, error) {
 	p, err := NewProducerFromClient(client)
 	if err != nil {
 		return nil, err
 	}
-	return newSyncProducerFromProducer(p), nil
+	return newSyncProducerFromProducer(p.(*producer)), nil
 }
 
-func newSyncProducerFromProducer(p *Producer) *SyncProducer {
+func newSyncProducerFromProducer(p *producer) *syncProducer {
 	p.conf.Producer.AckSuccesses = true
-	sp := &SyncProducer{producer: p}
+	sp := &syncProducer{producer: p}
 
 	sp.wg.Add(2)
 	go withRecover(sp.handleSuccesses)
@@ -39,9 +50,7 @@ func newSyncProducerFromProducer(p *Producer) *SyncProducer {
 	return sp
 }
 
-// SendMessage produces a message to the given topic with the given key and value. To send strings as either key or value, see the StringEncoder type.
-// It returns the partition and offset of the successfully-produced message, or the error (if any).
-func (sp *SyncProducer) SendMessage(topic string, key, value Encoder) (partition int32, offset int64, err error) {
+func (sp *syncProducer) SendMessage(topic string, key, value Encoder) (partition int32, offset int64, err error) {
 	expectation := make(chan error, 1)
 	msg := &ProducerMessage{Topic: topic, Key: key, Value: value, Metadata: expectation}
 	sp.producer.Input() <- msg
@@ -51,7 +60,7 @@ func (sp *SyncProducer) SendMessage(topic string, key, value Encoder) (partition
 	return
 }
 
-func (sp *SyncProducer) handleSuccesses() {
+func (sp *syncProducer) handleSuccesses() {
 	defer sp.wg.Done()
 	for msg := range sp.producer.Successes() {
 		expectation := msg.Metadata.(chan error)
@@ -59,7 +68,7 @@ func (sp *SyncProducer) handleSuccesses() {
 	}
 }
 
-func (sp *SyncProducer) handleErrors() {
+func (sp *syncProducer) handleErrors() {
 	defer sp.wg.Done()
 	for err := range sp.producer.Errors() {
 		expectation := err.Msg.Metadata.(chan error)
@@ -67,10 +76,7 @@ func (sp *SyncProducer) handleErrors() {
 	}
 }
 
-// Close shuts down the producer and flushes any messages it may have buffered. You must call this function before
-// a producer object passes out of scope, as it may otherwise leak memory. You must call this before calling Close
-// on the underlying client.
-func (sp *SyncProducer) Close() error {
+func (sp *syncProducer) Close() error {
 	sp.producer.AsyncClose()
 	sp.wg.Wait()
 	return nil