Add SASLMechanism config option with default to PLAIN

Author: myronjhicks

docs/config.md

@@ -72,6 +72,7 @@ ACHGateway:
       Topic: <string>
       TLS: <boolean>
       AutoCommit: <boolean>
+      [ SASLMechanism: <string> | default = "PLAIN" ]
       Transform:
         Encoding:
           [ Base64: <boolean> | default = false ]
@@ -156,6 +157,7 @@ ACHGateway:
         Topic: <string>
         [ TLS: <boolean> | default = false ]
         [ AutoCommit: <boolean> | default = false ]
+        [ SASLMechanism: <string> | default = "PLAIN" ]
     Webhook:
       [ Endpoint: <string> | default = "" ]
 ```

internal/kafka/kafka.go

@@ -21,7 +21,10 @@ func OpenTopic(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Topic, erro
 	config.Net.TLS.Enable = cfg.TLS
 
 	config.Net.SASL.Enable = cfg.Key != ""
-	config.Net.SASL.Mechanism = sarama.SASLMechanism("PLAIN")
+	config.Net.SASL.Mechanism = sarama.SASLMechanism(cfg.SASLMechanism)
+	if config.Net.SASL.Mechanism == "" {
+		config.Net.SASL.Mechanism = sarama.SASLMechanism("PLAIN")
+	}
 	config.Net.SASL.User = cfg.Key
 	config.Net.SASL.Password = cfg.Secret
 
@@ -46,7 +49,10 @@ func OpenSubscription(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Subs
 	config.Net.TLS.Enable = cfg.TLS
 
 	config.Net.SASL.Enable = cfg.Key != ""
-	config.Net.SASL.Mechanism = sarama.SASLMechanism("PLAIN")
+	config.Net.SASL.Mechanism = sarama.SASLMechanism(cfg.SASLMechanism)
+	if config.Net.SASL.Mechanism == "" {
+		config.Net.SASL.Mechanism = sarama.SASLMechanism("PLAIN")
+	}
 	config.Net.SASL.User = cfg.Key
 	config.Net.SASL.Password = cfg.Secret
 

internal/service/model_inbound.go

@@ -77,7 +77,8 @@ type KafkaConfig struct {
 	// AutoCommit in Sarama refers to "automated publishing of consumer offsets
 	// to the broker" rather than a Kafka broker's meaning of "commit consumer
 	// offsets on read" which leads to "at-most-once" delivery.
-	AutoCommit bool
+	AutoCommit    bool
+	SASLMechanism string
 
 	Consumer KafkaConsumerConfig
 	Producer KafkaProducerConfig