Add TLS configuration

consumerBuilder.go

@@ -341,6 +341,8 @@ func buildSaramaConfig(options *consumer.Options) *cluster.Config {
 	}
 	// TODO: make a cluster-wise kafkaVersion assignment. this hard coded assignment is used for enabling dlq consumer message timestamp injection.
 	config.Config.Version = sarama.V0_10_2_0
+	config.Config.Net.TLS.Enable = options.TLSConfig != nil
+	config.Config.Net.TLS.Config = options.TLSConfig
 	config.Config.Producer.MaxMessageBytes = options.ProducerMaxMessageByes
 	config.Config.Producer.RequiredAcks = sarama.WaitForAll
 	config.Config.Producer.Return.Successes = true

internal/consumer/options.go

@@ -21,6 +21,7 @@
 package consumer
 
 import (
+	"crypto/tls"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -42,6 +43,7 @@ type (
 		ProducerMaxMessageByes int
 		FetchDefaultBytes      int32
 		OtherConsumerTopics    []Topic
+		TLSConfig              *tls.Config // TLSConfig is the configuration to use for secure connections, not nil -> enable, nil -> disabled
 	}
 )
 

kafka/config.go

@@ -21,6 +21,7 @@
 package kafka
 
 import (
+	"crypto/tls"
 	"fmt"
 	"time"
 
@@ -94,6 +95,11 @@ type (
 		// When using the handler based API, this corresponds to the number of concurrent go
 		// 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
 	}
 )