🐛 bug(55827):[source-kafka]: added AWS MSK IAM Auth jar to classpath an… (#55828)

Co-authored-by: Marcos Marx <[email protected]>

Author: nickcopley-sa

airbyte-integrations/connectors/source-kafka/build.gradle

@@ -15,9 +15,11 @@ application {
 
 dependencies {
 
+    implementation 'software.amazon.msk:aws-msk-iam-auth:2.3.1'
     implementation 'org.apache.kafka:kafka-clients:3.2.1'
     implementation 'org.apache.kafka:connect-json:3.2.1'
     implementation 'io.confluent:kafka-avro-serializer:7.2.1'
 
+
     testImplementation 'org.testcontainers:kafka:1.19.4'
 }

airbyte-integrations/connectors/source-kafka/metadata.yaml

@@ -10,7 +10,7 @@ data:
     - suite: integrationTests
   connectorType: source
   definitionId: d917a47b-8537-4d0d-8c10-36a9928d4265
-  dockerImageTag: 0.3.0
+  dockerImageTag: 0.4.0
   dockerRepository: airbyte/source-kafka
   documentationUrl: https://docs.airbyte.com/integrations/sources/kafka
   githubIssueLabel: source-kafka

airbyte-integrations/connectors/source-kafka/src/main/java/io/airbyte/integrations/source/kafka/format/AbstractFormat.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.msk.auth.iam.IAMClientCallbackHandler;
 
 public abstract class AbstractFormat implements KafkaFormat {
 
@@ -85,6 +86,9 @@ private Map<String, Object> propertiesByProtocol(final JsonNode config) {
         if (saslMechanism.equals(OAuthBearerLoginModule.OAUTHBEARER_MECHANISM)) {
           builder.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, protocolConfig.get("oauthbearer_token_endpoint_url").asText());
           builder.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, OAuthBearerLoginCallbackHandler.class.getName());
+        } else if (saslMechanism.equals("AWS_MSK_IAM")) {
+          // IAMClientCallbackHandler
+          builder.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, IAMClientCallbackHandler.class.getName());
         }
       }
       default -> throw new RuntimeException("Unexpected Kafka protocol: " + Jsons.serialize(protocol));

airbyte-integrations/connectors/source-kafka/src/main/java/io/airbyte/integrations/source/kafka/format/JsonFormat.java

@@ -6,7 +6,9 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
+import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.util.AutoCloseableIterator;
 import io.airbyte.commons.util.AutoCloseableIterators;
 import io.airbyte.protocol.models.Field;
@@ -151,7 +153,9 @@ protected AirbyteMessage computeNext() {
               .withRecord(new AirbyteRecordMessage()
                   .withStream(record.topic())
                   .withEmittedAt(Instant.now().toEpochMilli())
-                  .withData(record.value()));
+                  .withData(record.value())
+                  .withData(Jsons.jsonNode(ImmutableMap.builder().put("value", record.value()).build())));
+
         }
 
         return endOfData();

airbyte-integrations/connectors/source-kafka/src/main/resources/spec.json

@@ -188,7 +188,8 @@
                   "OAUTHBEARER",
                   "SCRAM-SHA-256",
                   "SCRAM-SHA-512",
-                  "PLAIN"
+                  "PLAIN",
+                  "AWS_MSK_IAM"
                 ]
               },
               "sasl_jaas_config": {

docs/integrations/sources/kafka.md

@@ -68,6 +68,7 @@ AVRO - deserialize Using confluent API. Please refer (https://docs.confluent.io/
 
 | Version | Date       | Pull Request                                                                                       | Subject                                                              |
 | :------ | :--------- |:---------------------------------------------------------------------------------------------------|:---------------------------------------------------------------------|
+| 0.4.0   | 2025-03-18 | [55828](https://github.com/airbytehq/airbyte/pull/55828)                                           | Add configurations for AWS MSK IAM and fix JSON format               |
 | 0.3.0   | 2025-02-18 | [53231](https://github.com/airbytehq/airbyte/pull/53231)                                           | Add configurations for OAUTHBEARER SASL Mechanism                    |
 | 0.2.8 | 2025-02-07 | [53221](https://github.com/airbytehq/airbyte/pull/53221)                                           | For AVRO MessageFormat, schema_registry_password is a secret         |
 | 0.2.7 | 2025-01-10 | [51480](https://github.com/airbytehq/airbyte/pull/51480)                                           | Use a non root base image                                            |