Robert Walters

Senior Product Manager

MongoDB

Jagadish Nallapaneni

[AWS Identity and Access Management (IAM)](https://aws.amazon.com/iam/) is a widely used service that enables developers to control secure access to AWS services and resources. A key benefit of AWS IAM is its ability to make secure authenticated connections without the client having to pass sensitive passwords.

As of version 1.13, the MongoDB Connector for Apache Kafka (Kafka Connector) now supports authentication via AWS IAM. AWS IAM support has been one of the most commonly requested features for the Kafka Connector, and in this tutorial, we will walk through how to use it.

To achieve support for AWS IAM, we added support for overriding the MongoClient interface allowing a developer to provide a *CustomCredentialProvider*. While this provider can be used for any authentication method, this tutorial will focus on building an AWS IAM credential provider.

To illustrate this, we will walk through a simple configuration of an AWS EC2 instance installed with Apache Kafka. This instance will use the [MongoDB Connector for Apache Kafka](https://www.mongodb.com/docs/kafka-connector/current/) to [sink](https://en.wikipedia.org/wiki/Sink_(computing)) data from a Kafka topic to an Atlas cluster, and authenticate using AWS IAM.

## Prerequisites

If you wish to follow along in this article, make sure you have the following:

- An AWS IAM role, “KafkaBlogAtlasRole”. This role will later be mapped as an Atlas database user.
- MongoDB Atlas cluster. If you do not have an Atlas cluster, learn [how to get started with Atlas](https://www.mongodb.com/docs/atlas/getting-started/)..
- AWS EC2 instance
    - Install [Apache Kafka](https://kafka.apache.org/downloads).
    - Configure the EC2 instance with an AWS IAM role, “KafkaConnectBlogRole.”
    - Add a policy to KafkaConnectBlogRole that allows “AssumeRole” on the KafkaBlogAtlasRole.

Example policy:

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "KafkaAtlasBlogPolicy",
            "Effect": "Allow",
            "Action": "sts:AssumeRole",
            "Resource": "arn:aws:iam::979559056307:role/KafkaBlogAtlasRole"
        }
    ]
}
```


- Java Development Kit version 8 or later
- A database user in the desired MongoDB Atlas cluster

    - Configure the database user with AWS IAM and use the AWS Role ARN for KafkaBlogAtlasRole. The Atlas database user should have access to read/write to the desired collection.

![MongoDB Atlas edit user dialog showing AWS IAM authentication method][1]

At this point, we are ready to start building the custom authentication provider.

## Getting started

Starting with version 1.13 of the MongoDB Connector for Apache Kafka, we now have the ability to plug in a custom authentication provider. In this article, we will create a custom authentication provider in Java, compile it, and add it to the lib directory of your Kafka deployment. The provider we will create will support AWS IAM authentication. 

### Building the custom authentication component

To start, in your favorite Java editor, create a project and build the following JAR that contains this source:

SampleAssumeRoleCredential.java

```java
package com.mongodb;

import java.util.Map;
import java.util.function.Supplier;

import com.mongodb.kafka.connect.util.custom.credentials.CustomCredentialProvider;

import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceAsyncClientBuilder;
import com.amazonaws.services.securitytoken.model.AssumeRoleRequest;
import com.amazonaws.services.securitytoken.model.AssumeRoleResult;
import com.amazonaws.services.securitytoken.model.Credentials;
import com.amazonaws.util.StringUtils;

public class SampleAssumeRoleCredential implements CustomCredentialProvider {

  public SampleAssumeRoleCredential() {}
  @Override
  public MongoCredential getCustomCredential(Map<?, ?> map) {
    AWSCredentialsProvider provider = new DefaultAWSCredentialsProviderChain();
    Supplier<AwsCredential> awsFreshCredentialSupplier = () -> {
      AWSSecurityTokenService stsClient = AWSSecurityTokenServiceAsyncClientBuilder.standard()
          .withCredentials(provider)
          .withRegion("us-east-1")
          .build();
      AssumeRoleRequest assumeRoleRequest = new AssumeRoleRequest().withDurationSeconds(3600)
          .withRoleArn((String)map.get("mongodbaws.auth.mechanism.roleArn"))
          .withRoleSessionName("Test_Session");
      AssumeRoleResult assumeRoleResult = stsClient.assumeRole(assumeRoleRequest);
      Credentials creds = assumeRoleResult.getCredentials();
      // Add your code to fetch new credentials
      return new AwsCredential(creds.getAccessKeyId(), creds.getSecretAccessKey(), creds.getSessionToken());
    };
    return MongoCredential.createAwsCredential(null, null)
        .withMechanismProperty(MongoCredential.AWS_CREDENTIAL_PROVIDER_KEY, awsFreshCredentialSupplier);
  }

  @Override
  public void validate(Map<?, ?> map) {
    String roleArn = (String) map.get("mongodbaws.auth.mechanism.roleArn");
    if (StringUtils.isNullOrEmpty(roleArn)) {
      throw new RuntimeException("Invalid value set for customProperty");
    }
  }

  @Override
  public void init(Map<?, ?> map) {

  }
}
```

Example pom.xml file

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.mongodb</groupId>
    <artifactId>SampleAssumeRoleCredential</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.3</version>
                <configuration>
                    
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        
        
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver-sync</artifactId>
            <version>5.1.0</version>
        </dependency>
        
        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-java-sdk</artifactId>
            <version>1.12.723</version>
        </dependency>

        
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-jdk14</artifactId>
            <version>1.7.28</version>
        </dependency>

        <dependency>
            <groupId>kafka-connect</groupId>
            <artifactId>kafka-connect</artifactId>
            <scope>system</scope>
            <version>1.13.0</version>
        </dependency>
    </dependencies>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

</project>
```

Note: We need to reference MongoDB Connector for Apache Kafka version 1.13 or later. In this example, the connector was downloaded from the Kafka Connector GitHub repository and stored in the local Maven repository. 

```bash
mvn install:install-file -Dfile=/Users/robert.walters/Downloads/mongodb-kafka-connect-mongodb-1.13.0/lib/mongo-kafka-connect-1.13.0-confluent.jar -DgroupId=kafka-connect -DartifactId=kafka-connector -Dversion=1.13.0  -Dpackaging=jar\n
```

Compile and place this SampleAssumeRole.jar file in the Kafka libs directory in your Kafka Connect deployment. You are now ready to configure your MongoDB Connector for Apache Kafka to use this custom authentication provider.

### Configure the MongoDB Connector for Apache Kafka

To support the custom provider in version 1.13, the following configuration options were added to the connector:

| Configuration Parameter                   | Description                                                  |
| ----------------------------------------- | ------------------------------------------------------------ |
| mongo.custom.auth.mechanism.enable        | Determines if the connector should use the provider class; default is false |
| mongo.custom.auth.mechanism.providerClass | Java class entry point for the custom provider               |
| mongodbaws.auth.mechanism.roleArn         | Role that has IAM access to the database                     |

In this example, the custom authentication component will use AWS IAM AssumeRole to authenticate with MongoDB Atlas. Here is an example of this configuration connecting to a MongoDB Atlas cluster.

```json
{
    "name": "mongo-tutorial-sink",
    "config": {
        "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
        "topics": "Stocks2",
        "connection.uri": "<MongoDB Connection String>",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": false,
        "database": "Stocks",
        "collection": "StockData",
        "mongo.custom.auth.mechanism.enable":"true",
        "mongo.custom.auth.mechanism.providerClass":"com.mongodb.SampleAssumeRoleCredential",
        "mongodbaws.auth.mechanism.roleArn":"<AWS IAM ARN>"      
    }
}
```

Notice that the roleArn is the ARN (Amazon Resource Name) of the AWS IAM role that is used for Atlas users, KafkaBlogAtlasRole. At runtime, since Kafka Connect is running under the AWS IAM role of the EC2 instance, KafkaConnectBlogRole, and this role has a policy that allows AssumeRole on the KafkaBlogAtlasRole, the service can connect to Atlas without storing any credentials!

You are now ready to send data to Kafka and watch it arrive in a MongoDB collection using the MongoDB Connector for Apache Kafka configured with AWS IAM authentication to MongoDB Atlas.

## Summary

In this article, we discussed the requirements needed to support a custom authentication provider such as AWS IAM authentication. Version 1.13 of the Kafka Connector introduced new parameters that define which class to use within the custom provider JAR file. This JAR file includes the logic for the custom authentication. 

Note that while adding AWS IAM authentication capability natively to the Kafka Connector is possible (it is just software, after all), doing so would not only add a version dependency on the AWS IAM SDK, but it would also significantly increase the size of the connector, making it impractical for all other users who do not wish to use this authentication method. For these reasons, writing the custom credential provider is the solution for integrating the Kafka Connector with authentication methods such as AWS IAM.

Read the [MongoDB Connector for Apache Kafka documentation](https://www.mongodb.com/docs/kafka-connector/current/) to learn more. Questions? Join us in the [MongoDB Developer Community](https://www.mongodb.com/community/forums/).


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb44f9cef8df8f2b2/6682f292b1132488c03a422b/image1.png

To connect to an Atlas cluster, users authenticate with AWS IAM authentication. Learn how to set up the MongoDB Connector for Apache Kafka for authentication.

Jagadish Nallapaneni

Content by Jagadish Nallapaneni

Using AWS IAM Authentication with the MongoDB Connector for Apache Kafka