Add AWS MSK IAM support to Kafka sink YAML config

Allow users to define Kafka sinks with AWS MSK IAM authentication via
sequin.yaml by parsing/exporting the aws_access_key_id,
aws_secret_access_key, and aws_region fields.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: RTLS

lib/sequin/transforms/transforms.ex

@@ -223,7 +223,10 @@ defmodule Sequin.Transforms do
       tls: sink.tls,
       username: sink.username,
       password: SensitiveValue.new(sink.password, show_sensitive),
-      sasl_mechanism: sink.sasl_mechanism
+      sasl_mechanism: sink.sasl_mechanism,
+      aws_access_key_id: sink.aws_access_key_id,
+      aws_secret_access_key: SensitiveValue.new(sink.aws_secret_access_key, show_sensitive),
+      aws_region: sink.aws_region
     })
   end
 
@@ -1136,7 +1139,10 @@ defmodule Sequin.Transforms do
          tls: attrs["tls"] || false,
          username: attrs["username"],
          password: attrs["password"],
-         sasl_mechanism: sasl_mechanism
+         sasl_mechanism: sasl_mechanism,
+         aws_access_key_id: attrs["aws_access_key_id"],
+         aws_secret_access_key: attrs["aws_secret_access_key"],
+         aws_region: attrs["aws_region"]
        }}
     end
   end
@@ -1428,13 +1434,14 @@ defmodule Sequin.Transforms do
   defp parse_sasl_mechanism("plain"), do: {:ok, :plain}
   defp parse_sasl_mechanism("scram_sha_256"), do: {:ok, :scram_sha_256}
   defp parse_sasl_mechanism("scram_sha_512"), do: {:ok, :scram_sha_512}
+  defp parse_sasl_mechanism("aws_msk_iam"), do: {:ok, :aws_msk_iam}
 
   defp parse_sasl_mechanism(invalid),
     do:
       {:error,
        Error.validation(
          summary:
-           "[sasl_mechanism] Invalid SASL mechanism '#{invalid}'. Must be one of: plain, scram_sha_256, scram_sha_512"
+           "[sasl_mechanism] Invalid SASL mechanism '#{invalid}'. Must be one of: plain, scram_sha_256, scram_sha_512, aws_msk_iam"
        )}
 
   defp env do

test/sequin/yaml_loader_test.exs

@@ -1233,6 +1233,41 @@ defmodule Sequin.YamlLoaderTest do
              } = consumer.sink
     end
 
+    test "creates kafka sink consumer with AWS MSK IAM" do
+      assert :ok =
+               YamlLoader.apply_from_yml!("""
+               #{account_and_db_yml()}
+
+               sinks:
+                 - name: "kafka-msk-consumer"
+                   database: "test-db"
+                   destination:
+                     type: "kafka"
+                     hosts: "b-1.msk-cluster.abc123.kafka.us-east-1.amazonaws.com:9098"
+                     topic: "test-topic"
+                     tls: true
+                     sasl_mechanism: "aws_msk_iam"
+                     aws_access_key_id: "AKIAXXXXXXXXXXXXXXXX"
+                     aws_secret_access_key: "secret123"
+                     aws_region: "us-east-1"
+               """)
+
+      assert [consumer] = Repo.all(SinkConsumer)
+
+      assert consumer.name == "kafka-msk-consumer"
+
+      assert %KafkaSink{
+               type: :kafka,
+               hosts: "b-1.msk-cluster.abc123.kafka.us-east-1.amazonaws.com:9098",
+               topic: "test-topic",
+               tls: true,
+               sasl_mechanism: :aws_msk_iam,
+               aws_access_key_id: "AKIAXXXXXXXXXXXXXXXX",
+               aws_secret_access_key: "secret123",
+               aws_region: "us-east-1"
+             } = consumer.sink
+    end
+
     test "creates sqs sink consumer" do
       assert :ok =
                YamlLoader.apply_from_yml!("""