feat: Add kafka msk iam support

api/jsonschema/schema.json

@@ -213,6 +213,19 @@
       ],
       "type": "object"
     },
+    "io.argoproj.events.v1alpha1.AWSMSKIAMConfig": {
+      "description": "AWSMSKIAMConfig holds configuration for SASL/OAUTHBEARER authentication against Amazon MSK using IAM credentials (includes IRSA / web identity token support).",
+      "properties": {
+        "region": {
+          "description": "Region is the AWS region of the MSK cluster (e.g. \"us-east-1\").",
+          "type": "string"
+        }
+      },
+      "required": [
+        "region"
+      ],
+      "type": "object"
+    },
     "io.argoproj.events.v1alpha1.Amount": {
       "description": "Amount represent a numeric amount.",
       "type": "number"
@@ -2512,6 +2525,10 @@
     "io.argoproj.events.v1alpha1.KafkaEventSource": {
       "description": "KafkaEventSource refers to event-source for Kafka related events",
       "properties": {
+        "awsMskIamAuth": {
+          "$ref": "#/definitions/io.argoproj.events.v1alpha1.AWSMSKIAMConfig",
+          "description": "AWSMSKIAMAuth configures SASL/OAUTHBEARER authentication using AWS IAM credentials, supporting IRSA (pod web-identity token), instance-profile, and static env credentials. When set, TLS is enabled automatically and the SASL config field is ignored."
+        },
         "config": {
           "description": "Yaml format Sarama config for Kafka connection. It follows the struct of sarama.Config. See https://github.com/IBM/sarama/blob/main/config.go e.g.\n\nconsumer:\n  fetch:\n    min: 1\nnet:\n  MaxOpenRequests: 5",
           "type": "string"

docs/APIs.md

@@ -1126,6 +1126,72 @@ RoleARN is the Amazon Resource Name (ARN) of the role to assume.
 
 </table>
 
+<h3 id="argoproj.io/v1alpha1.AWSMSKIAMConfig">
+
+AWSMSKIAMConfig
+</h3>
+
+<p>
+
+(<em>Appears on:</em>
+<a href="#argoproj.io/v1alpha1.KafkaEventSource">KafkaEventSource</a>)
+</p>
+
+<p>
+
+<p>
+
+AWSMSKIAMConfig holds configuration for SASL/OAUTHBEARER authentication
+against Amazon MSK using IAM credentials (includes IRSA / web identity
+token support).
+</p>
+
+</p>
+
+<table>
+
+<thead>
+
+<tr>
+
+<th>
+
+Field
+</th>
+
+<th>
+
+Description
+</th>
+
+</tr>
+
+</thead>
+
+<tbody>
+
+<tr>
+
+<td>
+
+<code>region</code></br> <em> string </em>
+</td>
+
+<td>
+
+<p>
+
+Region is the AWS region of the MSK cluster (e.g. “us-east-1”).
+</p>
+
+</td>
+
+</tr>
+
+</tbody>
+
+</table>
+
 <h3 id="argoproj.io/v1alpha1.Amount">
 
 Amount
@@ -12439,6 +12505,30 @@ Schema Registry configuration for consumer message with Avro format
 
 </tr>
 
+<tr>
+
+<td>
+
+<code>awsMskIamAuth</code></br> <em>
+<a href="#argoproj.io/v1alpha1.AWSMSKIAMConfig"> AWSMSKIAMConfig </a>
+</em>
+</td>
+
+<td>
+
+<em>(Optional)</em>
+<p>
+
+AWSMSKIAMAuth configures SASL/OAUTHBEARER authentication using AWS IAM
+credentials, supporting IRSA (pod web-identity token), instance-profile,
+and static env credentials. When set, TLS is enabled automatically and
+the SASL config field is ignored.
+</p>
+
+</td>
+
+</tr>
+
 </tbody>
 
 </table>

docs/eventsources/setup/kafka.md

@@ -47,3 +47,74 @@ Kafka event-source specification is available [here](../../APIs.md#argoproj.io/v
 ## Troubleshoot
 
 Please read the [FAQ](https://argoproj.github.io/argo-events/FAQ/).
+
+## AWS MSK with IAM Authentication (IRSA)
+
+When connecting to Amazon MSK with IAM access control enabled, use the `awsMskIamAuth` field instead of `sasl` or `tls`.
+TLS is enabled automatically. The AWS credential chain is used, so the simplest way to grant access on EKS is via
+[IRSA (IAM Roles for Service Accounts)](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html).
+
+### IAM policy for the role
+
+The IAM role attached to the pod service account needs at minimum:
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": "kafka-cluster:Connect",
+      "Resource": "arn:aws:kafka:<region>:<account-id>:cluster/<cluster-name>/<cluster-id>"
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "kafka-cluster:DescribeTopic",
+        "kafka-cluster:ReadData"
+      ],
+      "Resource": "arn:aws:kafka:<region>:<account-id>:topic/<cluster-name>/<cluster-id>/<topic-name>"
+    },
+    {
+      "Effect": "Allow",
+      "Action": "kafka-cluster:AlterGroup",
+      "Resource": "arn:aws:kafka:<region>:<account-id>:group/<cluster-name>/<cluster-id>/<consumer-group-name>"
+    }
+  ]
+}
+```
+
+### EventSource spec
+
+```yaml
+apiVersion: argoproj.io/v1alpha1
+kind: EventSource
+metadata:
+  name: kafka-msk
+spec:
+  kafka:
+    msk-source:
+      # Use port 9098 for IAM authentication (not 9092)
+      url: b-1.my-cluster.abc123.c2.kafka.us-east-1.amazonaws.com:9098
+      topic: my-topic
+      consumerGroup:
+        groupName: my-consumer-group
+      awsMskIamAuth:
+        region: us-east-1
+```
+
+### Annotate the service account
+
+```yaml
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: argo-events-sa
+  namespace: argo-events
+  annotations:
+    eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/<role-name>
+```
+
+The `awsMskIamAuth` block uses the AWS SDK v2 default credential chain, so it also works with EC2 instance profiles
+and static environment credentials (`AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` / `AWS_SESSION_TOKEN`) without
+any additional configuration.

examples/event-sources/kafka.yaml

@@ -58,3 +58,15 @@ spec:
 #        userSecret:
 #          key: user
 #          name: my-user
+
+##    Enable AWS MSK IAM authentication (IRSA / web-identity token)
+##    Requires the pod service account to be annotated with an IAM role ARN:
+##      eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/<role-name>
+##    TLS is enabled automatically; do not also set the sasl or tls blocks.
+#  aws-msk-iam-example:
+#    url: b-1.my-cluster.abc123.c2.kafka.us-east-1.amazonaws.com:9098
+#    topic: my-topic
+#    consumerGroup:
+#      groupName: my-group
+#    awsMskIamAuth:
+#      region: us-east-1

go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/apache/openwhisk-client-go v0.0.0-20190915054138-716c6f973eb2
 	github.com/apache/pulsar-client-go v0.12.0
 	github.com/argoproj/notifications-engine v0.4.1-0.20250309174002-87bf0576a872
+	github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4
 	github.com/aws/aws-sdk-go v1.47.11
 	github.com/blushft/go-diagrams v0.0.0-20201006005127-c78c821223d9
 	github.com/bradleyfalzon/ghinstallation/v2 v2.18.0

go.sum

@@ -148,6 +148,8 @@ github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPd
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
 github.com/awalterschulze/gographviz v0.0.0-20200901124122-0eecad45bd71 h1:m3N1Fv5vE5IcxuTOGFGGV0grrVFHV8UY2SV0wSBXAC8=
 github.com/awalterschulze/gographviz v0.0.0-20200901124122-0eecad45bd71/go.mod h1:/ynarkO/43wP/JM2Okn61e8WFMtdbtA8he7GJxW+SFM=
+github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4 h1:2jAwFwA0Xgcx94dUId+K24yFabsKYDtAhCgyMit6OqE=
+github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4/go.mod h1:MVYeeOhILFFemC/XlYTClvBjYZrg/EPd3ts885KrNTI=
 github.com/aws/aws-sdk-go v1.32.6/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
 github.com/aws/aws-sdk-go v1.47.11 h1:Dol+MA+hQblbnXUI3Vk9qvoekU6O1uDEuAItezjiWNQ=
 github.com/aws/aws-sdk-go v1.47.11/go.mod h1:LF8svs817+Nz+DmiMQKTO3ubZ/6IaTpq3TjupRn3Eqk=

pkg/apis/events/v1alpha1/eventsource_types.go

@@ -498,6 +498,11 @@ type KafkaEventSource struct {
 	// Schema Registry configuration for consumer message with Avro format
 	// +optional
 	SchemaRegistry *SchemaRegistryConfig `json:"schemaRegistry,omitempty" protobuf:"bytes,14,opt,name=schemaRegistry"`
+	// AWSMSKIAMAuth configures SASL/OAUTHBEARER authentication using AWS IAM credentials,
+	// supporting IRSA (pod web-identity token), instance-profile, and static env credentials.
+	// When set, TLS is enabled automatically and the SASL config field is ignored.
+	// +optional
+	AWSMSKIAMAuth *AWSMSKIAMConfig `json:"awsMskIamAuth,omitempty" protobuf:"bytes,15,opt,name=awsMskIamAuth"`
 }
 
 type KafkaConsumerGroup struct {