add basic sasl_ssl support:

* add security_protocol to authConfig. a single value, SASL_SSL, is supported for now. this allows
  one to enable TLS support
* add sasl_mechanism to authConfig. currently we check for PLAIN.

common.go

@@ -164,6 +164,15 @@ func randomString(length int) string {
 	return fmt.Sprintf("%x", buf)[:length]
 }
 
+func parseSaslMechanism(auth authConfig) (sarama.SASLMechanism, error) {
+	switch strings.ToLower(auth.SASLMechanism) {
+	case "plain", "":
+		return sarama.SASLTypePlaintext, nil;
+	default:
+		return sarama.SASLTypePlaintext, fmt.Errorf("Unsupported auth sasl mechanism: %#v", auth.SASLMechanism)
+	}
+}
+
 // setupCerts takes the paths to a tls certificate, CA, and certificate key in
 // a PEM format and returns a constructed tls.Config object.
 func setupCerts(certPath, caPath, keyPath string) (*tls.Config, error) {
@@ -207,6 +216,8 @@ type authConfig struct {
 	ClientCertKey     string `json:"client-certificate-key"`
 	SASLPlainUser     string `json:"sasl_plain_user"`
 	SASLPlainPassword string `json:"sasl_plain_password"`
+	SASLMechanism     string `json:"sasl_mechanism"`
+	SecurityProtocol  string `json:"security_protocol"`
 }
 
 func setupAuth(auth authConfig, saramaCfg *sarama.Config) error {
@@ -230,6 +241,16 @@ func setupSASL(auth authConfig, saramaCfg *sarama.Config) error {
 	saramaCfg.Net.SASL.Enable = true
 	saramaCfg.Net.SASL.User = auth.SASLPlainUser
 	saramaCfg.Net.SASL.Password = auth.SASLPlainPassword
+	saslMechanism, err := parseSaslMechanism(auth)
+
+	if err != nil {
+		return err;
+	}
+	saramaCfg.Net.SASL.Mechanism = saslMechanism;
+
+	if (strings.EqualFold(auth.SecurityProtocol, "SASL_SSL")) {
+		saramaCfg.Net.TLS.Enable = true
+	}
 	return nil
 }