Repository: beatlabs/patron
Pull request #197 (2020-04-07 08:09:31)

Add support for Kafka synchronous producer through builder
This PR is a proposed solution for #36.

## Short description of the changes

* allow creating a synchronous Kafka producer through a specific builder option

I have separated the `SyncProducer` and `AsyncProducer`, while keeping a minimal `Producer` interface.

The unit test `TestSendWithCustomEncoder` now uses the synchronous producer to simplify the error handling.

Fixes as well the description of the metrics which was copied from the consumer.

This PR supersedes #170. 

gm42: The CI build fails with `client has run out of available brokers to talk to (Is your cluster reachable?)` for the synchronous message deliveries; I wonder if we need to wait a bit in `docker-compose` before attempting to reach the Kafka cluster in the tests.
mantzas: > The CI build fails with `client has run out of available brokers to talk to (Is your cluster reachable?)` for the synchronous message deliveries; I wonder if we need to wait a bit in `docker-compose` before attempting to reach the Kafka cluster in the tests.

Kafka is not in docker-compose... sarama gives you a way to start a in-memory broker...
gm42: > Kafka is not in docker-compose... sarama gives you a way to start a in-memory broker...

Sure; then I need to investigate why this happens.
gm42: For the records, I am inquiring upstream about how to use the MockBroker with v0.11.0+ (for headers support we need for the messages): https://github.com/Shopify/sarama/issues/1665

It might take a while though and require a PR upstream...
mantzas: > For the records, I am inquiring upstream about how to use the MockBroker with v0.11.0+ (for headers support we need for the messages): [Shopify/sarama#1665](https://github.com/Shopify/sarama/issues/1665)
> 
> It might take a while though and require a PR upstream...

Maybe we could introduce actual kafka and integration test and solve it the correct way. Mockbroker is not 100% Kafka and we will always have such problems...
gm42: > Maybe we could introduce actual kafka and integration test and solve it the correct way. Mockbroker is not 100% Kafka and we will always have such problems...

Agree with you, just did not want to introduce too big changes for this PR. I can try to add it in the `docker-compose` if you like.
mantzas: > > Maybe we could introduce actual kafka and integration test and solve it the correct way. Mockbroker is not 100% Kafka and we will always have such problems...
> 
> Agree with you, just did not want to introduce too big changes for this PR. I can try to add it in the `docker-compose` if you like.

Issue #199 and we should do it with dockertest since we have decided to use dockertest for that.
When #199 is done and merged you should be ok.
mantzas: @gm42 the CI seems to be broken...
gm42: @mantzas that hasn't changed, it's `failed to create producer client: kafka: client has run out of available brokers to talk to (Is your cluster reachable?)`. See https://github.com/beatlabs/patron/pull/197#issuecomment-615105190
mantzas: I am approving. Tests will be handled with proper Kafka integration tests #212 

Patch:
diff --git a/client/kafka/async_producer.go b/client/kafka/async_producer.go
new file mode 100644
index 000000000..67b1c1a2f
--- /dev/null
+++ b/client/kafka/async_producer.go
@@ -0,0 +1,65 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/beatlabs/patron/trace"
+
+	"github.com/Shopify/sarama"
+	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
+)
+
+// AsyncProducer is an asynchronous Kafka producer.
+type AsyncProducer struct {
+	baseProducer
+	asyncProd sarama.AsyncProducer
+	chErr     chan error
+}
+
+// Send a message to a topic, asynchronously. Producer errors are queued on the
+// channel obtained during the AsyncProducer creation.
+func (ap *AsyncProducer) Send(ctx context.Context, msg *Message) error {
+	sp, _ := trace.ChildSpan(ctx, trace.ComponentOpName(asyncProducerComponent, msg.topic),
+		asyncProducerComponent, ext.SpanKindProducer, ap.tag,
+		opentracing.Tag{Key: "topic", Value: msg.topic})
+	pm, err := ap.createProducerMessage(ctx, msg, sp)
+	if err != nil {
+		ap.statusCountInc(messageCreationErrors, msg.topic)
+		trace.SpanError(sp)
+		return err
+	}
+
+	ap.statusCountInc(messageSent, msg.topic)
+	ap.asyncProd.Input() <- pm
+	trace.SpanSuccess(sp)
+
+	return nil
+}
+
+func (ap *AsyncProducer) propagateError() {
+	for pe := range ap.asyncProd.Errors() {
+		ap.statusCountInc(messageSendErrors, pe.Msg.Topic)
+		ap.chErr <- fmt.Errorf("failed to send message: %w", pe)
+	}
+}
+
+// Close shuts down the producer and waits for any buffered messages to be
+// flushed. You must call this function before a producer object passes out of
+// scope, as it may otherwise leak memory.
+func (ap *AsyncProducer) Close() error {
+	err := ap.asyncProd.Close()
+	if err != nil {
+		// always close client
+		_ = ap.prodClient.Close()
+
+		return fmt.Errorf("failed to close async producer client: %w", err)
+	}
+
+	err = ap.prodClient.Close()
+	if err != nil {
+		return fmt.Errorf("failed to close async producer: %w", err)
+	}
+	return nil
+}
diff --git a/client/kafka/builder.go b/client/kafka/builder.go
index 2d66b790b..f9f6c2221 100644
--- a/client/kafka/builder.go
+++ b/client/kafka/builder.go
@@ -5,12 +5,13 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/Shopify/sarama"
 	"github.com/beatlabs/patron/encoding"
 	"github.com/beatlabs/patron/encoding/json"
 	patronErrors "github.com/beatlabs/patron/errors"
 	"github.com/beatlabs/patron/internal/validation"
 	"github.com/beatlabs/patron/log"
+
+	"github.com/Shopify/sarama"
 	"github.com/opentracing/opentracing-go"
 )
 
@@ -29,22 +30,20 @@ const (
 
 const fieldSetMsg = "Setting property '%v' for '%v'"
 
-// AsyncBuilder gathers all required and optional properties, in order
-// to construct a Kafka AsyncProducer.
-type AsyncBuilder struct {
+// Builder gathers all required and optional properties, in order
+// to construct a Kafka AsyncProducer/SyncProducer.
+type Builder struct {
 	brokers     []string
 	cfg         *sarama.Config
-	chErr       chan error
-	tag         opentracing.Tag
 	enc         encoding.EncodeFunc
 	contentType string
 	errors      []error
 }
 
-// NewBuilder initiates the AsyncProducer builder chain.
+// NewBuilder initiates the AsyncProducer/SyncProducer builder chain.
 // The builder instantiates the component using default values for
 // EncodeFunc and Content-Type header.
-func NewBuilder(brokers []string) *AsyncBuilder {
+func NewBuilder(brokers []string) *Builder {
 	cfg := sarama.NewConfig()
 	cfg.Version = sarama.V0_11_0_0
 
@@ -53,19 +52,17 @@ func NewBuilder(brokers []string) *AsyncBuilder {
 		errs = append(errs, errors.New("brokers are empty or have an empty value"))
 	}
 
-	return &AsyncBuilder{
+	return &Builder{
 		brokers:     brokers,
 		cfg:         cfg,
-		chErr:       make(chan error),
-		tag:         opentracing.Tag{Key: "type", Value: "async"},
 		enc:         json.Encode,
 		contentType: json.Type,
 		errors:      errs,
 	}
 }
 
-// WithTimeout sets the dial timeout for the AsyncProducer.
-func (ab *AsyncBuilder) WithTimeout(dial time.Duration) *AsyncBuilder {
+// WithTimeout sets the dial timeout for the sync or async producer.
+func (ab *Builder) WithTimeout(dial time.Duration) *Builder {
 	if dial <= 0 {
 		ab.errors = append(ab.errors, errors.New("dial timeout has to be positive"))
 		return ab
@@ -75,8 +72,8 @@ func (ab *AsyncBuilder) WithTimeout(dial time.Duration) *AsyncBuilder {
 	return ab
 }
 
-// WithVersion sets the kafka versionfor the AsyncProducer.
-func (ab *AsyncBuilder) WithVersion(version string) *AsyncBuilder {
+// WithVersion sets the kafka versionfor the AsyncProducer/SyncProducer.
+func (ab *Builder) WithVersion(version string) *Builder {
 	if version == "" {
 		ab.errors = append(ab.errors, errors.New("version is required"))
 		return ab
@@ -94,7 +91,7 @@ func (ab *AsyncBuilder) WithVersion(version string) *AsyncBuilder {
 
 // WithRequiredAcksPolicy adjusts how many replica acknowledgements
 // broker must see before responding.
-func (ab *AsyncBuilder) WithRequiredAcksPolicy(ack RequiredAcks) *AsyncBuilder {
+func (ab *Builder) WithRequiredAcksPolicy(ack RequiredAcks) *Builder {
 	if !isValidRequiredAcks(ack) {
 		ab.errors = append(ab.errors, errors.New("invalid value for required acks policy provided"))
 		return ab
@@ -106,7 +103,7 @@ func (ab *AsyncBuilder) WithRequiredAcksPolicy(ack RequiredAcks) *AsyncBuilder {
 
 // WithEncoder sets a specific encoder implementation and Content-Type string header;
 // if no option is provided it defaults to json.
-func (ab *AsyncBuilder) WithEncoder(enc encoding.EncodeFunc, contentType string) *AsyncBuilder {
+func (ab *Builder) WithEncoder(enc encoding.EncodeFunc, contentType string) *Builder {
 	if enc == nil {
 		ab.errors = append(ab.errors, errors.New("encoder is nil"))
 	} else {
@@ -123,34 +120,73 @@ func (ab *AsyncBuilder) WithEncoder(enc encoding.EncodeFunc, contentType string)
 	return ab
 }
 
-// Create constructs the AsyncProducer component by applying the gathered properties.
-func (ab *AsyncBuilder) Create() (*AsyncProducer, error) {
+// CreateAsync constructs the AsyncProducer component by applying the gathered properties.
+func (ab *Builder) CreateAsync() (*AsyncProducer, <-chan error, error) {
 
 	if len(ab.errors) > 0 {
-		return nil, patronErrors.Aggregate(ab.errors...)
+		return nil, nil, patronErrors.Aggregate(ab.errors...)
 	}
 
-	prodClient, err := sarama.NewClient(ab.brokers, ab.cfg)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create async producer client: %w", err)
+	ap := AsyncProducer{
+		baseProducer: baseProducer{
+			messageStatus: messageStatus,
+			deliveryType:  "async",
+			cfg:           ab.cfg,
+			enc:           ab.enc,
+			contentType:   ab.contentType,
+			tag:           opentracing.Tag{Key: "type", Value: "async"},
+		},
 	}
-	prod, err := sarama.NewAsyncProducerFromClient(prodClient)
+
+	var err error
+	ap.prodClient, err = sarama.NewClient(ab.brokers, ab.cfg)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create async producer: %w", err)
+		return nil, nil, fmt.Errorf("failed to create producer client: %w", err)
 	}
 
-	ap := AsyncProducer{
-		cfg:         ab.cfg,
-		prodClient:  prodClient,
-		prod:        prod,
-		chErr:       ab.chErr,
-		enc:         ab.enc,
-		contentType: ab.contentType,
-		tag:         ab.tag,
+	ap.asyncProd, err = sarama.NewAsyncProducerFromClient(ap.prodClient)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create async producer: %w", err)
 	}
+	ap.chErr = make(chan error)
 
 	go ap.propagateError()
-	return &ap, nil
+
+	return &ap, ap.chErr, nil
+}
+
+// CreateSync constructs the SyncProducer component by applying the gathered properties.
+func (ab *Builder) CreateSync() (*SyncProducer, error) {
+	if len(ab.errors) > 0 {
+		return nil, patronErrors.Aggregate(ab.errors...)
+	}
+
+	// required for any SyncProducer; 'Errors' is already true by default for both async/sync producers
+	ab.cfg.Producer.Return.Successes = true
+
+	p := SyncProducer{
+		baseProducer: baseProducer{
+			messageStatus: messageStatus,
+			deliveryType:  "sync",
+			cfg:           ab.cfg,
+			enc:           ab.enc,
+			contentType:   ab.contentType,
+			tag:           opentracing.Tag{Key: "type", Value: "sync"},
+		},
+	}
+
+	var err error
+	p.prodClient, err = sarama.NewClient(ab.brokers, ab.cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create producer client: %w", err)
+	}
+
+	p.syncProd, err = sarama.NewSyncProducerFromClient(p.prodClient)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create sync producer: %w", err)
+	}
+
+	return &p, nil
 }
 
 func isValidRequiredAcks(ack RequiredAcks) bool {
diff --git a/client/kafka/kafka.go b/client/kafka/kafka.go
index 31645b9ab..1a65fc993 100644
--- a/client/kafka/kafka.go
+++ b/client/kafka/kafka.go
@@ -5,40 +5,48 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/Shopify/sarama"
 	"github.com/beatlabs/patron/correlation"
 	"github.com/beatlabs/patron/encoding"
-	"github.com/beatlabs/patron/trace"
+
+	"github.com/Shopify/sarama"
 	"github.com/opentracing/opentracing-go"
-	"github.com/opentracing/opentracing-go/ext"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
 const (
-	producerComponent     = "kafka-async-producer"
-	messageCreationErrors = "creation-errors"
-	messageSendErrors     = "send-errors"
-	messageSent           = "sent"
+	asyncProducerComponent = "kafka-async-producer"
+	syncProducerComponent  = "kafka-sync-producer"
+	messageCreationErrors  = "creation-errors"
+	messageSendErrors      = "send-errors"
+	messageSent            = "sent"
 )
 
-var messageStatus *prometheus.CounterVec
+var (
+	messageStatus *prometheus.CounterVec
+)
 
-func messageStatusCountInc(status, topic string) {
-	messageStatus.WithLabelValues(status, topic).Inc()
+// Producer interface for Kafka.
+type Producer interface {
+	Send(ctx context.Context, msg *Message) error
+	Close() error
 }
 
-func init() {
-	messageStatus = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Namespace: "component",
-			Subsystem: "kafka_async_producer",
-			Name:      "message_status",
-			Help:      "Message status counter (received, decoded, decoding-errors) classified by topic",
-		}, []string{"status", "topic"},
-	)
-	prometheus.MustRegister(messageStatus)
+type baseProducer struct {
+	cfg         *sarama.Config
+	prodClient  sarama.Client
+	tag         opentracing.Tag
+	enc         encoding.EncodeFunc
+	contentType string
+	// deliveryType can be 'sync' or 'async'
+	deliveryType  string
+	messageStatus *prometheus.CounterVec
 }
 
+var (
+	_ Producer = &AsyncProducer{}
+	_ Producer = &SyncProducer{}
+)
+
 // Message abstraction of a Kafka message.
 type Message struct {
 	topic string
@@ -46,6 +54,19 @@ type Message struct {
 	key   *string
 }
 
+func init() {
+	messageStatus = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "component",
+			Subsystem: "kafka_producer",
+			Name:      "message_status",
+			Help:      "Message status counter (produced, encoded, encoding-errors) classified by topic",
+		}, []string{"status", "topic", "type"},
+	)
+
+	prometheus.MustRegister(messageStatus)
+}
+
 // NewMessage creates a new message.
 func NewMessage(t string, b interface{}) *Message {
 	return &Message{topic: t, body: b}
@@ -59,49 +80,13 @@ func NewMessageWithKey(t string, b interface{}, k string) (*Message, error) {
 	return &Message{topic: t, body: b, key: &k}, nil
 }
 
-// Producer interface for Kafka.
-type Producer interface {
-	Send(ctx context.Context, msg *Message) error
-	Error() <-chan error
-	Close() error
-}
-
-// AsyncProducer defines a async Kafka producer.
-type AsyncProducer struct {
-	cfg         *sarama.Config
-	prodClient  sarama.Client
-	prod        sarama.AsyncProducer
-	chErr       chan error
-	tag         opentracing.Tag
-	enc         encoding.EncodeFunc
-	contentType string
-}
-
-// Send a message to a topic.
-func (ap *AsyncProducer) Send(ctx context.Context, msg *Message) error {
-	sp, _ := trace.ChildSpan(ctx, trace.ComponentOpName(producerComponent, msg.topic),
-		producerComponent, ext.SpanKindProducer, ap.tag,
-		opentracing.Tag{Key: "topic", Value: msg.topic})
-	pm, err := ap.createProducerMessage(ctx, msg, sp)
-	if err != nil {
-		messageStatusCountInc(messageCreationErrors, msg.topic)
-		trace.SpanError(sp)
-		return err
-	}
-	messageStatusCountInc(messageSent, msg.topic)
-	ap.prod.Input() <- pm
-	trace.SpanSuccess(sp)
-	return nil
-}
-
-// Error returns a chanel to monitor for errors.
-func (ap *AsyncProducer) Error() <-chan error {
-	return ap.chErr
+func (p *baseProducer) statusCountInc(status, topic string) {
+	p.messageStatus.WithLabelValues(status, topic, p.deliveryType).Inc()
 }
 
 // ActiveBrokers returns a list of active brokers' addresses.
-func (ap *AsyncProducer) ActiveBrokers() []string {
-	brokers := ap.prodClient.Brokers()
+func (p *baseProducer) ActiveBrokers() []string {
+	brokers := p.prodClient.Brokers()
 	activeBrokerAddresses := make([]string, len(brokers))
 	for i, b := range brokers {
 		activeBrokerAddresses[i] = b.Addr()
@@ -109,46 +94,20 @@ func (ap *AsyncProducer) ActiveBrokers() []string {
 	return activeBrokerAddresses
 }
 
-// Close shuts down the producer and waits for any buffered messages to be
-// flushed. You must call this function before a producer object passes out of
-// scope, as it may otherwise leak memory.
-func (ap *AsyncProducer) Close() error {
-	err := ap.prod.Close()
-	if err != nil {
-		// always close client
-		_ = ap.prodClient.Close()
-
-		return fmt.Errorf("failed to close async producer client: %w", err)
-	}
-
-	err = ap.prodClient.Close()
-	if err != nil {
-		return fmt.Errorf("failed to close async producer: %w", err)
-	}
-	return nil
-}
-
-func (ap *AsyncProducer) propagateError() {
-	for pe := range ap.prod.Errors() {
-		messageStatusCountInc(messageSendErrors, pe.Msg.Topic)
-		ap.chErr <- fmt.Errorf("failed to send message: %w", pe)
-	}
-}
-
-func (ap *AsyncProducer) createProducerMessage(ctx context.Context, msg *Message, sp opentracing.Span) (*sarama.ProducerMessage, error) {
+func (p *baseProducer) createProducerMessage(ctx context.Context, msg *Message, sp opentracing.Span) (*sarama.ProducerMessage, error) {
 	c := kafkaHeadersCarrier{}
 	err := sp.Tracer().Inject(sp.Context(), opentracing.TextMap, &c)
 	if err != nil {
 		return nil, fmt.Errorf("failed to inject tracing headers: %w", err)
 	}
-	c.Set(encoding.ContentTypeHeader, ap.contentType)
+	c.Set(encoding.ContentTypeHeader, p.contentType)
 
 	var saramaKey sarama.Encoder
 	if msg.key != nil {
 		saramaKey = sarama.StringEncoder(*msg.key)
 	}
 
-	b, err := ap.enc(msg.body)
+	b, err := p.enc(msg.body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to encode message body: %w", err)
 	}
diff --git a/client/kafka/sync_producer.go b/client/kafka/sync_producer.go
new file mode 100644
index 000000000..c7d610b33
--- /dev/null
+++ b/client/kafka/sync_producer.go
@@ -0,0 +1,63 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/beatlabs/patron/trace"
+
+	"github.com/Shopify/sarama"
+	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
+)
+
+// SyncProducer is a synchronous Kafka producer.
+type SyncProducer struct {
+	baseProducer
+
+	syncProd sarama.SyncProducer
+}
+
+// Send a message to a topic.
+func (p *SyncProducer) Send(ctx context.Context, msg *Message) error {
+	sp, _ := trace.ChildSpan(ctx, trace.ComponentOpName(syncProducerComponent, msg.topic),
+		syncProducerComponent, ext.SpanKindProducer, p.tag,
+		opentracing.Tag{Key: "topic", Value: msg.topic})
+	pm, err := p.createProducerMessage(ctx, msg, sp)
+	if err != nil {
+		p.statusCountInc(messageCreationErrors, msg.topic)
+		trace.SpanError(sp)
+		return err
+	}
+
+	_, _, err = p.syncProd.SendMessage(pm)
+	if err != nil {
+		p.statusCountInc(messageCreationErrors, msg.topic)
+		trace.SpanError(sp)
+		return err
+	}
+
+	p.statusCountInc(messageSent, msg.topic)
+	trace.SpanSuccess(sp)
+
+	return nil
+}
+
+// Close shuts down the producer and waits for any buffered messages to be
+// flushed. You must call this function before a producer object passes out of
+// scope, as it may otherwise leak memory.
+func (p *SyncProducer) Close() error {
+	err := p.syncProd.Close()
+	if err != nil {
+		// always close client
+		_ = p.prodClient.Close()
+
+		return fmt.Errorf("failed to close sync producer client: %w", err)
+	}
+
+	err = p.prodClient.Close()
+	if err != nil {
+		return fmt.Errorf("failed to close sync producer: %w", err)
+	}
+	return nil
+}
diff --git a/examples/second/main.go b/examples/second/main.go
index 904ace7f6..58f306a5d 100644
--- a/examples/second/main.go
+++ b/examples/second/main.go
@@ -76,10 +76,16 @@ type httpComponent struct {
 }
 
 func newHTTPComponent(kafkaBroker, topic, url string) (*httpComponent, error) {
-	prd, err := kafka.NewBuilder([]string{kafkaBroker}).Create()
+	prd, chErr, err := kafka.NewBuilder([]string{kafkaBroker}).CreateAsync()
 	if err != nil {
 		return nil, err
 	}
+	go func() {
+		for {
+			err := <-chErr
+			log.Errorf("error producing Kafka message: %v", err)
+		}
+	}()
 	return &httpComponent{prd: prd, topic: topic}, nil
 }
 


Test patch:
diff --git a/client/kafka/builder_test.go b/client/kafka/builder_test.go
index fc1bbb8eb..c25b39f45 100644
--- a/client/kafka/builder_test.go
+++ b/client/kafka/builder_test.go
@@ -5,10 +5,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Shopify/sarama"
 	"github.com/beatlabs/patron/encoding"
 	"github.com/beatlabs/patron/encoding/json"
 	"github.com/beatlabs/patron/encoding/protobuf"
+
+	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -199,12 +200,12 @@ func Test_createAsyncProducerUsingBuilder(t *testing.T) {
 
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
-			gotAsyncProducer, gotErrs := NewBuilder(tt.brokers).
+			gotAsyncProducer, chErr, gotErrs := NewBuilder(tt.brokers).
 				WithVersion(tt.version).
 				WithRequiredAcksPolicy(tt.ack).
 				WithTimeout(tt.timeout).
 				WithEncoder(tt.enc, tt.contentType).
-				Create()
+				CreateAsync()
 
 			v, _ := sarama.ParseKafkaVersion(tt.version)
 			if len(tt.wantErrs) > 0 {
@@ -212,6 +213,7 @@ func Test_createAsyncProducerUsingBuilder(t *testing.T) {
 				assert.Nil(t, gotAsyncProducer)
 			} else {
 				assert.NotNil(t, gotAsyncProducer)
+				assert.NotNil(t, chErr)
 				assert.IsType(t, &AsyncProducer{}, gotAsyncProducer)
 				assert.EqualValues(t, v, gotAsyncProducer.cfg.Version)
 				assert.EqualValues(t, tt.ack, gotAsyncProducer.cfg.Producer.RequiredAcks)
diff --git a/client/kafka/kafka_test.go b/client/kafka/kafka_test.go
index 6365528d6..84ef04fb0 100644
--- a/client/kafka/kafka_test.go
+++ b/client/kafka/kafka_test.go
@@ -4,12 +4,13 @@ import (
 	"context"
 	"testing"
 
-	"github.com/Shopify/sarama"
 	"github.com/beatlabs/patron/encoding"
 	"github.com/beatlabs/patron/encoding/json"
 	"github.com/beatlabs/patron/encoding/protobuf"
 	"github.com/beatlabs/patron/examples"
 	"github.com/beatlabs/patron/trace"
+
+	"github.com/Shopify/sarama"
 	"github.com/prometheus/client_golang/prometheus"
 	io_prometheus_client "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
@@ -19,7 +20,7 @@ import (
 type testMetric struct {
 	metric *prometheus.CounterVec
 	name   string
-	label  string
+	labels []string
 	count  uint64
 }
 
@@ -53,51 +54,93 @@ func TestNewMessageWithKey(t *testing.T) {
 	}
 }
 
+func TestNewAsyncProducer_Failure(t *testing.T) {
+	got, chErr, err := NewBuilder([]string{}).CreateAsync()
+	assert.Error(t, err)
+	assert.Nil(t, got)
+	assert.Nil(t, chErr)
+}
+
+func TestNewAsyncProducer_Option_Failure(t *testing.T) {
+	got, chErr, err := NewBuilder([]string{"xxx"}).WithVersion("xxxx").CreateAsync()
+	assert.Error(t, err)
+	assert.Nil(t, got)
+	assert.Nil(t, chErr)
+}
+
+func TestNewAsyncProducer_Success(t *testing.T) {
+	seed := createKafkaBroker(t, false)
+	got, chErr, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).CreateAsync()
+	assert.NoError(t, err)
+	assert.NotNil(t, got)
+	assert.NotNil(t, chErr)
+}
+
 func TestNewSyncProducer_Failure(t *testing.T) {
-	got, err := NewBuilder([]string{}).Create()
+	got, err := NewBuilder([]string{}).CreateSync()
 	assert.Error(t, err)
 	assert.Nil(t, got)
 }
 
 func TestNewSyncProducer_Option_Failure(t *testing.T) {
-	got, err := NewBuilder([]string{"xxx"}).WithVersion("xxxx").Create()
+	got, err := NewBuilder([]string{"xxx"}).WithVersion("xxxx").CreateSync()
 	assert.Error(t, err)
 	assert.Nil(t, got)
 }
 
 func TestNewSyncProducer_Success(t *testing.T) {
 	seed := createKafkaBroker(t, false)
-	got, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).Create()
+	got, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).CreateSync()
 	assert.NoError(t, err)
 	assert.NotNil(t, got)
 }
 
 func TestAsyncProducer_SendMessage_Close(t *testing.T) {
 	msg := NewMessage("TOPIC", "TEST")
-	tm := testMetric{messageStatus, "component_kafka_async_producer_message_status", "sent", 1}
+	tm := testMetric{messageStatus, "component_kafka_async_producer_message_status", []string{"sent", "async"}, 1}
 	seed := createKafkaBroker(t, true)
-	ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).Create()
+	ap, chErr, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).CreateAsync()
 	assert.NoError(t, err)
 	assert.NotNil(t, ap)
+	assert.NotNil(t, chErr)
 	err = trace.Setup("test", "1.0.0", "0.0.0.0:6831", jaeger.SamplerTypeProbabilistic, 0.1)
 	assert.NoError(t, err)
 	_, ctx := trace.ChildSpan(context.Background(), "123", "cmp")
 	clearMetrics(tm)
 	err = ap.Send(ctx, msg)
-	assertMetric(t, tm)
 	assert.NoError(t, err)
-	assert.Error(t, <-ap.Error())
+	assertMetric(t, tm)
+	assert.Error(t, <-chErr)
 	assert.NoError(t, ap.Close())
 }
 
+func TestSyncProducer_SendMessage_Close(t *testing.T) {
+	msg := NewMessage("TOPIC", "TEST")
+	tm := testMetric{messageStatus, "component_kafka_sync_producer_message_status", []string{"sent", "sync"}, 1}
+	seed := createKafkaBroker(t, true)
+	p, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_11_0_0.String()).CreateSync()
+	if !assert.NoError(t, err) {
+		return
+	}
+	assert.NotNil(t, p)
+	err = trace.Setup("test", "1.0.0", "0.0.0.0:6831", jaeger.SamplerTypeProbabilistic, 0.1)
+	assert.NoError(t, err)
+	_, ctx := trace.ChildSpan(context.Background(), "123", "cmp")
+	clearMetrics(tm)
+	err = p.Send(ctx, msg)
+	assert.NoError(t, err)
+	assertMetric(t, tm)
+	assert.NoError(t, p.Close())
+}
+
 func TestAsyncProducer_SendMessage_WithKey(t *testing.T) {
 	testKey := "TEST"
 	msg, err := NewMessageWithKey("TOPIC", "TEST", testKey)
-	tm := testMetric{messageStatus, "component_kafka_async_producer_message_status", "sent", 1}
+	tm := testMetric{messageStatus, "component_kafka_async_producer_message_status", []string{"sent", "async"}, 1}
 	assert.Equal(t, testKey, *msg.key)
 	assert.NoError(t, err)
 	seed := createKafkaBroker(t, true)
-	ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).Create()
+	ap, chErr, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).CreateAsync()
 	assert.NoError(t, err)
 	assert.NotNil(t, ap)
 	err = trace.Setup("test", "1.0.0", "0.0.0.0:6831", jaeger.SamplerTypeProbabilistic, 0.1)
@@ -105,9 +148,31 @@ func TestAsyncProducer_SendMessage_WithKey(t *testing.T) {
 	_, ctx := trace.ChildSpan(context.Background(), "123", "cmp")
 	clearMetrics(tm)
 	err = ap.Send(ctx, msg)
+	assert.NoError(t, err)
 	assertMetric(t, tm)
+	assert.Error(t, <-chErr)
+	assert.NoError(t, ap.Close())
+}
+
+func TestSyncProducer_SendMessage_WithKey(t *testing.T) {
+	testKey := "TEST"
+	msg, err := NewMessageWithKey("TOPIC", "TEST", testKey)
+	tm := testMetric{messageStatus, "component_kafka_sync_producer_message_status", []string{"sent", "sync"}, 1}
+	assert.Equal(t, testKey, *msg.key)
+	assert.NoError(t, err)
+	seed := createKafkaBroker(t, true)
+	ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_11_0_0.String()).CreateSync()
+	if !assert.NoError(t, err) {
+		return
+	}
+	assert.NotNil(t, ap)
+	err = trace.Setup("test", "1.0.0", "0.0.0.0:6831", jaeger.SamplerTypeProbabilistic, 0.1)
+	assert.NoError(t, err)
+	_, ctx := trace.ChildSpan(context.Background(), "123", "cmp")
+	clearMetrics(tm)
+	err = ap.Send(ctx, msg)
 	assert.NoError(t, err)
-	assert.Error(t, <-ap.Error())
+	assertMetric(t, tm)
 	assert.NoError(t, ap.Close())
 }
 
@@ -147,13 +212,13 @@ func TestSendWithCustomEncoder(t *testing.T) {
 		tm          []testMetric
 		wantSendErr bool
 	}{
-		{name: "json success", data: "testdata1", key: "testkey1", enc: json.Encode, ct: json.Type, tm: []testMetric{{messageStatus, "component_kafka_async_producer_message_status", "sent", 1}}, wantSendErr: false},
-		{name: "protobuf success", data: &u, key: "testkey2", enc: protobuf.Encode, ct: protobuf.Type, tm: []testMetric{{messageStatus, "component_kafka_async_producer_message_status", "sent", 1}}, wantSendErr: false},
+		{name: "json success", data: "testdata1", key: "testkey1", enc: json.Encode, ct: json.Type, tm: []testMetric{{messageStatus, "component_kafka_sync_producer_message_status", []string{"sent", "sync"}, 1}}, wantSendErr: false},
+		{name: "protobuf success", data: &u, key: "testkey2", enc: protobuf.Encode, ct: protobuf.Type, tm: []testMetric{{messageStatus, "component_kafka_sync_producer_message_status", []string{"sent", "sync"}, 1}}, wantSendErr: false},
 		{name: "failure due to invalid data", data: make(chan bool), key: "testkey3", wantSendErr: true},
 		{name: "nil message data", data: nil, key: "testkey4", wantSendErr: false},
 		{name: "nil encoder", data: "somedata", key: "testkey5", ct: json.Type, wantSendErr: false},
-		{name: "empty data", data: "", key: "testkey6", enc: json.Encode, ct: json.Type, tm: []testMetric{{messageStatus, "component_kafka_async_producer_message_status", "sent", 1}}, wantSendErr: false},
-		{name: "empty data two", data: "", key: "🚖", enc: json.Encode, ct: json.Type, tm: []testMetric{{messageStatus, "component_kafka_async_producer_message_status", "sent", 1}}, wantSendErr: false},
+		{name: "empty data", data: "", key: "testkey6", enc: json.Encode, ct: json.Type, tm: []testMetric{{messageStatus, "component_kafka_sync_producer_message_status", []string{"sent", "sync"}, 1}}, wantSendErr: false},
+		{name: "empty data two", data: "", key: "🚖", enc: json.Encode, ct: json.Type, tm: []testMetric{{messageStatus, "component_kafka_sync_producer_message_status", []string{"sent", "sync"}, 1}}, wantSendErr: false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -161,7 +226,7 @@ func TestSendWithCustomEncoder(t *testing.T) {
 			msg, _ := NewMessageWithKey("TOPIC", tt.data, tt.key)
 
 			seed := createKafkaBroker(t, true)
-			ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).WithEncoder(tt.enc, tt.ct).Create()
+			ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_11_0_0.String()).WithEncoder(tt.enc, tt.ct).CreateSync()
 			if tt.enc != nil {
 				assert.NoError(t, err)
 			} else {
@@ -199,18 +264,36 @@ func assertMetric(t *testing.T, testMetrics ...testMetric) {
 	assert.Len(t, metricFamilies, len(testMetrics))
 
 	var current *io_prometheus_client.Metric
+	found := map[string]struct{}{}
+	expected := map[string]struct{}{}
 	// Loop over our test metrics
 	for _, v := range testMetrics {
-		// And find the one which matches the label
+		for _, l := range v.labels {
+			expected[l] = struct{}{}
+		}
+
+		// And find the one which matches the labels
 		for _, mf := range metricFamilies {
 			for _, m := range mf.Metric {
 				for _, l := range m.Label {
-					if *l.Value == v.label {
-						current = m
+					found[*l.Value] = struct{}{}
+
+					for _, lbl := range v.labels {
+						if *l.Value == lbl {
+							current = m
+							break
+						}
 					}
 				}
 			}
 		}
+		// will fail in case of metric mismatch e.g. `creation-errors` instead of `sent`
+		assert.NotNil(t, current)
+		if current == nil {
+			t.Errorf("found: %v expected: %v", found, expected)
+			continue
+		}
+
 		// Then, perform the assertions on the matched counter
 		counter := current.Counter
 		if v.count > 0 {
@@ -227,7 +310,19 @@ func assertMetric(t *testing.T, testMetrics ...testMetric) {
 
 func TestAsyncProducerActiveBrokers(t *testing.T) {
 	seed := createKafkaBroker(t, true)
-	ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).Create()
+	ap, chErr, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).CreateAsync()
+	assert.NoError(t, err)
+	assert.NotNil(t, ap)
+	assert.NotNil(t, chErr)
+
+	assert.NotEmpty(t, ap.ActiveBrokers())
+
+	assert.NoError(t, ap.Close())
+}
+
+func TestSyncProducerActiveBrokers(t *testing.T) {
+	seed := createKafkaBroker(t, true)
+	ap, err := NewBuilder([]string{seed.Addr()}).WithVersion(sarama.V0_8_2_0.String()).CreateSync()
 	assert.NoError(t, err)
 	assert.NotNil(t, ap)