feat(kafka): add iam authentication (#1265)

formancehq · Feb 29, 2024 · 2ce9717 · 2ce9717
1 parent ce2f827
commit 2ce9717
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 11 deletions.
diff --git a/benthos/config.yml b/benthos/config.yml
@@ -1,3 +1,3 @@
 http:
   enabled: true
-  address: 0.0.0.0:4195
+  address: 0.0.0.0:4195
diff --git a/benthos/templates/event_bus.yaml b/benthos/templates/event_bus.yaml
@@ -9,21 +9,31 @@ fields:
 
 mapping: |
   root = if env("BROKER") == "kafka" {{
-    "kafka": {
-      "addresses": [ env("KAFKA_ADDRESS") ],
+    "kafka_franz": {
+      "seed_brokers": [ env("KAFKA_ADDRESS") ],
       "topics": [ env("TOPIC_PREFIX") + this.topic ],
-      "target_version": env("KAFKA_VERSION"),
       "consumer_group": this.consumer_group,
       "checkpoint_limit": 1024,
-      "sasl": {
-        "mechanism": env("KAFKA_SASL_MECHANISM"),
-        "password": env("KAFKA_SASL_PASSWORD"),
-        "user": env("KAFKA_SASL_USERNAME")
-      },
+      "sasl": [
+        {
+          "mechanism": env("KAFKA_SASL_MECHANISM"),
+          "password": env("KAFKA_SASL_PASSWORD"),
+          "username": env("KAFKA_SASL_USERNAME"),
+          "aws": {
+            "region": env("AWS_REGION"),
+            "credentials": {
+              "profile": env("AWS_PROFILE"),
+              "id": env("AWS_ACCESS_KEY_ID"),
+              "secret": env("AWS_SECRET_ACCESS_KEY"),
+              "token": env("AWS_SESSION_TOKEN"),
+              "role": env("AWS_ROLE_ARN")
+            }
+          }
+        }
+      ],
       "tls": {
         "enabled": env("KAFKA_TLS_ENABLED") == "true"
-      },
-      "extract_tracing_map": "root = meta()"
+      }
     }
   }} else {{
     "nats_jetstream": {