Add support for SASL plaintext auth

.travis.yml

@@ -2,7 +2,7 @@ language: go
 sudo: false
 go:
   - 1.11
-go_import_path: github.com/uber-go/kafka-client
+go_import_path: github.com/gig/kafka-client
 cache:
   directories:
     - vendor

Makefile

@@ -1,9 +1,9 @@
 export GO15VENDOREXPERIMENT=1
 
 BENCH_FLAGS ?= -cpuprofile=cpu.pprof -memprofile=mem.pprof -benchmem
-PKGS ?= ./cmd/... ./internal/... ./kafka/... .
+PKGS ?= ./cmd/... ./lib/... ./kafka/... .
 # Many Go tools take file globs or directories as arguments instead of packages.
-PKG_FILES ?= *.go kafka internal/consumer internal/backoff internal/list internal/metrics internal/util
+PKG_FILES ?= *.go kafka lib/consumer lib/backoff lib/list lib/metrics lib/util
 
 # The linting tools evolve with each Go version, so run them only on the latest
 # stable release.

README.md

@@ -12,11 +12,11 @@ A high level Go client library for Apache Kafka that provides the following prim
 This library is in alpha. APIs are subject to change, use at your own risk
 
 ## Contributing
-If you are interested in contributing, please sign the [License Agreement](https://cla-assistant.io/uber-go/kafka-client) and see our [development guide](https://github.com/uber-go/kafka-client/blob/master/docs/DEVELOPMENT-GUIDE.md)
+If you are interested in contributing, please sign the [License Agreement](https://cla-assistant.io/uber-go/kafka-client) and see our [development guide](https://github.com/gig/kafka-client/blob/master/docs/DEVELOPMENT-GUIDE.md)
 
 ## Installation
 
-`go get -u github.com/uber-go/kafka-client`
+`go get -u github.com/gig/kafka-client`
 
 ## Quick Start
 
@@ -27,8 +27,8 @@ import (
 	"os"
 	"os/signal"
 
-	"github.com/uber-go/kafka-client"
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )
@@ -100,7 +100,7 @@ func main() {
 ```
 
 [mit-img]: http://img.shields.io/badge/License-MIT-blue.svg
-[mit]: https://github.com/uber-go/kafka-client/blob/master/LICENSE
+[mit]: https://github.com/gig/kafka-client/blob/master/LICENSE
 
 [ci-img]: https://img.shields.io/travis/uber-go/kafka-client/master.svg
 [ci]: https://travis-ci.org/uber-go/kafka-client/branches

client.go

@@ -23,7 +23,7 @@ package kafkaclient
 import (
 	"os"
 
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )

client_test.go

@@ -27,8 +27,8 @@ import (
 	"github.com/Shopify/sarama"
 	cluster "github.com/bsm/sarama-cluster"
 	"github.com/stretchr/testify/assert"
-	"github.com/uber-go/kafka-client/internal/consumer"
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client/lib/consumer"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )

consumerBuilder.go

@@ -21,16 +21,21 @@
 package kafkaclient
 
 import (
+	"crypto/sha512"
+	"hash"
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/bsm/sarama-cluster"
-	"github.com/uber-go/kafka-client/internal/consumer"
-	"github.com/uber-go/kafka-client/kafka"
+	cluster "github.com/bsm/sarama-cluster"
+	"github.com/gig/kafka-client/lib/consumer"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
+	"github.com/xdg/scram"
 	"go.uber.org/zap"
 )
 
+var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
+
 type (
 	consumerBuilder struct {
 		clusterTopicsMap map[consumerCluster][]consumer.Topic
@@ -63,6 +68,12 @@ type (
 	}
 )
 
+type XDGSCRAMClient struct {
+	*scram.Client
+	*scram.ClientConversation
+	scram.HashGeneratorFcn
+}
+
 func newConsumerBuilder(
 	config *kafka.ConsumerConfig,
 	resolver kafka.NameResolver,
@@ -76,9 +87,9 @@ func newConsumerBuilder(
 		clusterSaramaClientMap:        make(map[consumer.ClusterGroup]sarama.Client),
 		clusterSaramaConsumerMap:      make(map[consumer.ClusterGroup]consumer.SaramaConsumer),
 		clusterTopicSaramaProducerMap: make(map[string]map[string]sarama.AsyncProducer),
-		msgCh:  make(chan kafka.Message, consumerOptions.RcvBufferSize),
-		logger: logger.With(zap.String("consumergroup", config.GroupName)),
-		scope:  scope.Tagged(map[string]string{"consumergroup": config.GroupName}),
+		msgCh:                         make(chan kafka.Message, consumerOptions.RcvBufferSize),
+		logger:                        logger.With(zap.String("consumergroup", config.GroupName)),
+		scope:                         scope.Tagged(map[string]string{"consumergroup": config.GroupName}),
 		constructors: consumer.Constructors{
 			NewSaramaConsumer: consumer.NewSaramaConsumer,
 			NewSaramaProducer: consumer.NewSaramaProducer,
@@ -124,6 +135,7 @@ func (c *consumerBuilder) Build() (kafka.Consumer, error) {
 
 func (c *consumerBuilder) build() (*consumer.MultiClusterConsumer, error) {
 	// build TopicList per cluster
+	errorsC := make(chan error)
 	for _, consumerTopic := range c.kafkaConfig.TopicList {
 		// first, add TopicConsumer for original topic if topic is well defined.
 		// disabling offset commit only applies for the original topic.
@@ -191,6 +203,7 @@ func (c *consumerBuilder) build() (*consumer.MultiClusterConsumer, error) {
 			Group:   cluster.groupName,
 		}] = consumer.NewClusterConsumer(
 			cluster.name,
+			errorsC,
 			saramaConsumer,
 			topicConsumerMap,
 			c.scope,
@@ -204,6 +217,7 @@ func (c *consumerBuilder) build() (*consumer.MultiClusterConsumer, error) {
 		c.kafkaConfig.TopicList,
 		clusterConsumerMap,
 		c.clusterSaramaClientMap,
+		errorsC,
 		c.msgCh,
 		c.scope,
 		c.logger,
@@ -325,6 +339,9 @@ func buildOptions(config *kafka.ConsumerConfig, consumerOpts ...ConsumerOption)
 		opts.OffsetPolicy = config.Offsets.Initial.Offset
 	}
 
+	opts.TLSConfig = config.TLSConfig
+	opts.SASLEnabled = config.SASLEnabled
+
 	// Apply optional consumer parameters that may be passed in.
 	for _, cOpt := range consumerOpts {
 		cOpt.apply(opts)
@@ -358,9 +375,38 @@ func buildSaramaConfig(options *consumer.Options) *cluster.Config {
 	config.Consumer.Offsets.CommitInterval = options.OffsetCommitInterval
 	config.Consumer.Offsets.Initial = options.OffsetPolicy
 	config.Consumer.MaxProcessingTime = options.MaxProcessingTime
+
+	if options.SASLEnabled {
+		config.Net.SASL.Enable = true
+		config.Net.SASL.User = options.SASLUsername
+		config.Net.SASL.Password = options.SASLPassword
+		config.Net.SASL.Handshake = true
+		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
+		config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
+		config.Net.TLS.Enable = true
+		config.Net.TLS.Config = options.TLSConfig
+	}
 	return config
 }
 
+func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) {
+	x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
+	if err != nil {
+		return err
+	}
+	x.ClientConversation = x.Client.NewConversation()
+	return nil
+}
+
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+	response, err = x.ClientConversation.Step(challenge)
+	return
+}
+
+func (x *XDGSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
+}
+
 func topicToRetryTopic(topic kafka.ConsumerTopic) kafka.ConsumerTopic {
 	return kafka.ConsumerTopic{
 		Topic:      topic.RetryQ,

consumerBuilder_test.go

@@ -28,8 +28,8 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/bsm/sarama-cluster"
 	"github.com/stretchr/testify/suite"
-	"github.com/uber-go/kafka-client/internal/consumer"
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client/lib/consumer"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )

consumerOptions.go

@@ -21,8 +21,8 @@
 package kafkaclient
 
 import (
-	"github.com/uber-go/kafka-client/internal/consumer"
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client/lib/consumer"
+	"github.com/gig/kafka-client/kafka"
 )
 
 type (
@@ -42,8 +42,26 @@ type (
 	clientIDOptions struct {
 		clientID string
 	}
+
+	SASLOptions struct {
+		username string
+		password string
+	}
 )
 
+func WithSASLAuth(username, password string) ConsumerOption {
+	return &SASLOptions{
+		username: username,
+		password: password,
+	}
+}
+
+func (o *SASLOptions) apply(opts *consumer.Options) {
+	opts.SASLEnabled = true
+	opts.SASLUsername = o.username
+	opts.SASLPassword = o.password
+}
+
 // WithDLQTopics creates a range consumer for the specified consumer DLQ topics.
 func WithDLQTopics(topicList kafka.ConsumerTopicList) ConsumerOption {
 	return &dlqTopicsOptions{

consumerOptions_test.go

@@ -24,8 +24,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/uber-go/kafka-client/internal/consumer"
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client/lib/consumer"
+	"github.com/gig/kafka-client/kafka"
 )
 
 func TestDLQConsumerOptions(t *testing.T) {

glide.yaml

@@ -1,4 +1,4 @@
-package: github.com/uber-go/kafka-client
+package: github.com/gig/kafka-client
 license: MIT
 import:
 - package: go.uber.org/zap

kafka/config.go

@@ -96,10 +96,11 @@ type (
 		// routines handler functions the library will run. Default is 1.
 		Concurrency int
 
-
 		// TLSConfig is the configuration to use for secure connections if
 		// enabled (not nil) (defaults to disabled, nil).
 		TLSConfig *tls.Config
+
+		SASLEnabled bool
 	}
 )
 

kafka/interfaces.go

@@ -37,6 +37,8 @@ type (
 		Start() error
 		// Stop stops the consumer
 		Stop()
+		// Errors returns errors from the sarama cluster
+		Errors() <-chan error
 		// Closed returns a channel which will be closed after this consumer is completely shutdown
 		Closed() <-chan struct{}
 		// Messages return the message channel for this consumer

internal/consumer/ackMgr.go → lib/consumer/ackMgr.go

@@ -25,8 +25,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/uber-go/kafka-client/internal/list"
-	"github.com/uber-go/kafka-client/internal/metrics"
+	"github.com/gig/kafka-client/lib/list"
+	"github.com/gig/kafka-client/lib/metrics"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )

internal/consumer/clusterConsumer.go → lib/consumer/clusterConsumer.go

@@ -27,10 +27,10 @@ import (
 	"sync"
 	"time"
 
-	"github.com/bsm/sarama-cluster"
-	"github.com/uber-go/kafka-client/internal/metrics"
-	"github.com/uber-go/kafka-client/internal/util"
-	"github.com/uber-go/kafka-client/kafka"
+	cluster "github.com/bsm/sarama-cluster"
+	"github.com/gig/kafka-client/lib/metrics"
+	"github.com/gig/kafka-client/lib/util"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -52,6 +52,7 @@ type (
 		metricsTicker    *time.Ticker
 		stopC            chan struct{}
 		doneC            chan struct{}
+		errorsC          chan error
 	}
 
 	ClusterGroup struct {
@@ -63,6 +64,7 @@ type (
 // NewClusterConsumer returns a new single cluster consumer.
 func NewClusterConsumer(
 	cluster string,
+	errorsC chan error,
 	saramaConsumer SaramaConsumer,
 	consumerMap map[string]*TopicConsumer,
 	scope tally.Scope,
@@ -78,6 +80,7 @@ func NewClusterConsumer(
 		metricsTicker:    time.NewTicker(metricsInterval),
 		stopC:            make(chan struct{}),
 		doneC:            make(chan struct{}),
+		errorsC:          errorsC,
 	}
 }
 
@@ -124,6 +127,11 @@ func (c *ClusterConsumer) Closed() <-chan struct{} {
 	return c.doneC
 }
 
+// Errors returns errors from the sarama cluster
+func (c *ClusterConsumer) Errors() <-chan error {
+	return c.errorsC
+}
+
 // eventLoop is the main event loop for this consumer
 func (c *ClusterConsumer) eventLoop() {
 	var n *cluster.Notification
@@ -141,6 +149,7 @@ func (c *ClusterConsumer) eventLoop() {
 		case err, ok := <-c.consumer.Errors():
 			if ok {
 				c.logger.Warn("cluster consumer error", zap.Error(err))
+				c.errorsC <- err
 			}
 		case _, ok := <-c.metricsTicker.C:
 			if ok && n != nil {

internal/consumer/clusterConsumer_test.go → lib/consumer/clusterConsumer_test.go

@@ -28,8 +28,8 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/bsm/sarama-cluster"
 	"github.com/stretchr/testify/suite"
-	"github.com/uber-go/kafka-client/internal/util"
-	"github.com/uber-go/kafka-client/kafka"
+	"github.com/gig/kafka-client/lib/util"
+	"github.com/gig/kafka-client/kafka"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )