use sub object within kafka for cloud provider specific settings related to kafka

Author: myronjhicks

docs/config.md

@@ -73,10 +73,11 @@ ACHGateway:
       TLS: <boolean>
       AutoCommit: <boolean>
       [ SASLMechanism: <string> | default = "PLAIN" ]
-      [ AWSRegion: <string> | default = "" ]
-      [ AWSProfile: <string> | default = "" ]
-      [ AWSRoleARN: <string> | default = "" ]
-      [ AWSSessionName: <string> | default = "" ]
+      AWS:
+        [ Region: <string> | default = "" ]
+        [ Profile: <string> | default = "" ]
+        [ RoleARN: <string> | default = "" ]
+        [ SessionName: <string> | default = "" ]
       Transform:
         Encoding:
           [ Base64: <boolean> | default = false ]
@@ -162,10 +163,11 @@ ACHGateway:
         [ TLS: <boolean> | default = false ]
         [ AutoCommit: <boolean> | default = false ]
         [ SASLMechanism: <string> | default = "PLAIN" ]
-        [ AWSRegion: <string> | default = "" ]
-        [ AWSProfile: <string> | default = "" ]
-        [ AWSRoleARN: <string> | default = "" ]
-        [ AWSSessionName: <string> | default = "" ]
+        AWS:
+          [ Region: <string> | default = "" ]
+          [ Profile: <string> | default = "" ]
+          [ RoleARN: <string> | default = "" ]
+          [ SessionName: <string> | default = "" ]
     Webhook:
       [ Endpoint: <string> | default = "" ]
 ```

internal/kafka/kafka.go

@@ -54,13 +54,15 @@ func OpenTopic(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Topic, erro
 
 	switch cfg.SASLMechanism {
 	case "AWS_MSK_IAM":
-		config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
-		config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{
-			Region:      cfg.AWSRegion,
-			Profile:     cfg.AWSProfile,
-			RoleARN:     cfg.AWSRoleARN,
-			SessionName: cfg.AWSSessionName,
+		if cfg.AWS != nil {
+			config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{
+				Region:      cfg.AWS.Region,
+				Profile:     cfg.AWS.Profile,
+				RoleARN:     cfg.AWS.RoleARN,
+				SessionName: cfg.AWS.SessionName,
+			}
 		}
+		config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
 		config.Net.TLS.Enable = true
 		config.Net.TLS.Config = &tls.Config{}
 
@@ -82,7 +84,6 @@ func OpenTopic(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Topic, erro
 		Set("sasl.mechanism", log.String(string(config.Net.SASL.Mechanism))).
 		Set("sasl.user", log.String(cfg.Key)).
 		Set("topic", log.String(cfg.Topic)).
-		Set("aws.region", log.String(cfg.AWSRegion)).
 		Log("opening kafka topic")
 
 	return kafkapubsub.OpenTopic(cfg.Brokers, config, cfg.Topic, nil)
@@ -97,13 +98,15 @@ func OpenSubscription(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Subs
 	// Default to PLAIN if no SASL mechanism is specified
 	switch cfg.SASLMechanism {
 	case "AWS_MSK_IAM":
-		config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
-		config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{
-			Region:      cfg.AWSRegion,
-			Profile:     cfg.AWSProfile,
-			RoleARN:     cfg.AWSRoleARN,
-			SessionName: cfg.AWSSessionName,
+		if cfg.AWS != nil {
+			config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{
+				Region:      cfg.AWS.Region,
+				Profile:     cfg.AWS.Profile,
+				RoleARN:     cfg.AWS.RoleARN,
+				SessionName: cfg.AWS.SessionName,
+			}
 		}
+		config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
 		config.Net.TLS.Enable = true
 		config.Net.TLS.Config = &tls.Config{}
 
@@ -129,7 +132,6 @@ func OpenSubscription(logger log.Logger, cfg *service.KafkaConfig) (*pubsub.Subs
 		Set("sasl.mechanism", log.String(string(config.Net.SASL.Mechanism))).
 		Set("sasl.user", log.String(cfg.Key)).
 		Set("topic", log.String(cfg.Topic)).
-		Set("aws.region", log.String(cfg.AWSRegion)).
 		Log("setting up kafka subscription")
 
 	return kafkapubsub.OpenSubscription(cfg.Brokers, config, cfg.Group, []string{cfg.Topic}, &kafkapubsub.SubscriptionOptions{

internal/service/model_inbound.go

@@ -66,6 +66,15 @@ func (cfg *InMemory) Validate() error {
 	return nil
 }
 
+// AWSConfig holds authentication settings for AWS MSK clusters.
+// Used when KafkaConfig.SASLMechanism is set to AWS_MSK_IAM.
+type AWSConfig struct {
+	Region      string
+	Profile     string
+	RoleARN     string
+	SessionName string
+}
+
 type KafkaConfig struct {
 	Brokers []string
 	Key     string
@@ -80,10 +89,7 @@ type KafkaConfig struct {
 	AutoCommit    bool
 	SASLMechanism string
 
-	AWSRegion      string
-	AWSProfile     string
-	AWSRoleARN     string
-	AWSSessionName string
+	AWS *AWSConfig
 
 	Consumer KafkaConsumerConfig
 	Producer KafkaProducerConfig