> For the complete MongoDB documentation index, see www.mongodb.com/docs/llms.txt

<!--
Tab options on this page. Append to the .md URL to filter:
  ?tabs=<id,...>   select specific tabs (e.g. ?tabs=nodejs,shell)
  ?allTabs=true    include every tab
  (no param)       default: one tab per tabset

Available tabs:
  other tabs: kafka-producer-param, kafka-consumer-param, stream-connect-kafka-public-ip, stream-connect-kafka-vpc-peering, stream-connect-kafka-plain-auth, stream-connect-kafka-ssl-auth, stream-connect-kafka-mtls-auth
-->

# Add an Atlas Stream Processing Connection

## Connection Types

Atlas Stream Processing supports a variety of [connection types](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections-considerations). The following procedures describe how to create these connections with standard security features.

### Add a Kafka Connection through the Atlas Administration API

The Atlas Administration API provides the [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) endpoint for adding a connection to a connection registry.

**Important:**

After adding an external connection such as an [Apache Kafka](https://kafka.apache.org/) cluster to your connection registry, you must add Atlas IP addresses to an access list for that external connection. To learn more, see [Allow Access to or from the Atlas Control Plane.](https://www.mongodb.com/docs/atlas/setup-cluster-security.md#std-label-atlas-add-inbound-ips)

#### Parameters Configuration

Each interface for creating a Kafka connection allows you to provide configuration parameters for your Kafka cluster. These configurations take the form of key-value pairs, and correspond to one of the following:

- A subset of [Producer](https://kafka.apache.org/documentation/#producerconfigs) configurations

- A subset of [Consumer](https://kafka.apache.org/documentation/#consumerconfigs) configurations.

Atlas Stream Processing passes only these parameters to your Kafka cluster. If you declare any parameters not explicitly allowed, Atlas Stream Processing ignores them.

| Interface | Configuration Mechanism |
| --- | --- |
| Atlas CLI | Provide configurations as a `.json` file specified with the `--file` flag. |
| Atlas Administration API | Provide configurations as a `.json` object specified in the `config` field. |
| Atlas UI | Provide configurations in the Configuration File field of the Add Connection page. |

Atlas Stream Processing supports the following configurations:

### Producer Parameters

- [acks](https://kafka.apache.org/documentation/#producerconfigs_acks)

- [batch.size](https://kafka.apache.org/documentation/#producerconfigs_batch.size)

- [client.dns.lookup](https://kafka.apache.org/documentation/#connectconfigs_client.dns.lookup)

- [client.id](https://kafka.apache.org/documentation/#consumerconfigs_client.id)

- [compression.type](https://kafka.apache.org/documentation/#producerconfigs_compression.type)

- [connections.max.idle.ms](https://kafka.apache.org/documentation/#brokerconfigs_connections.max.idle.ms)

- [delivery.timeout.ms](https://kafka.apache.org/documentation/#producerconfigs_delivery.timeout.ms)

- [enable.idempotence](https://kafka.apache.org/documentation/#producerconfigs_enable.idempotence)

- [linger.ms](https://kafka.apache.org/documentation/#producerconfigs_linger.ms)

- [max.in.flight.requests.per.connection](https://kafka.apache.org/documentation/#producerconfigs_max.in.flight.requests.per.connection)

- [message.max.bytes](https://kafka.apache.org/documentation/#brokerconfigs_message.max.bytes)

- [queue.buffering.max.messages](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md)

- [request.timeout.ms](https://kafka.apache.org/documentation/#producerconfigs_request.timeout.ms)

- [retries](https://kafka.apache.org/documentation/#producerconfigs_retries)

- [transactional.id](https://kafka.apache.org/documentation/#producerconfigs_transactional.id)

#### Supported Aggregation Behavior

You can use [Apache Kafka](https://kafka.apache.org/) connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stages.

### Add an Atlas Cluster Connection through the Atlas Administration API

The Atlas Administration API provides the [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) endpoint for adding a connection to a connection registry.

IMPORTANT: To create a cross-project Atlas connection, first enable the functionality. To learn more, see [Cross-Project Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections-cross-project)

#### Supported Aggregation Behavior

You can use Atlas cluster connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$merge](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-merge.md#std-label-atlas-sp-agg-merge) stages.

### Add a Kinesis Connection through the Atlas Administration API

To add a Kinesis connection to your stream processing workspace through the Atlas Administration API, follow these steps:

1. Set up Unified AWS Access.

   Follow the procedure described in [Set Up Unified AWS Access.](https://docs.mongodb.com/atlas/security/set-up-unified-aws-access/?interface=atlas-admin-api)

   Ensure that you grant your IAM role the following permissions:

   - `SubscribeToShard`

   - `ListShards`

   - `DescribeStreamSummary`

   - `PutRecords`

   Note the ARN value in `Statement.Principal.AWS` for later in this procedure.

2. In Atlas, go to the Users page.

   If it's not already displayed, select your desired organization from the  Organizations menu in the navigation bar.

   Click All Projects in the sidebar under the Identity & Access section, and select your desired project.

   Click Project Identity & Access in the sidebar under the Security section.

   The [Users](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2Faccess) page displays.

3. Create an API key.

   The [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) API endpoint requires digest authorization when creating an S3 (Simple Storage Service) connection. To support this, you must create an API Key.

   In the Project Access Manager, select the Applications tab, then click API Keys.

   Click Create API Key. Provide a short description for the key.

   In the Project Permissions dropdown menu, select both the Project Stream Processing Owner and Project Owner roles. Click Next.

   Save both the public and private keys for later in this procedure.

4. Create the Kinesis Connection.

   The Atlas Administration API provides an endpoint to [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/group/endpoint-/Streams/operation/createStreamConnection). You must send this request using digest authorization.

   In your HTTP request interface, enable digest authorization. For the `username`, provide the public key you generated previously. For the `password`, provide the private key you generated previously.

   For an AWS (Amazon Web Services) Kinesis connection, set the following key-value pairs:

   | Key | Value |
   | --- | --- |
   | `type` | `"AWSKinesisDataStreams"` |
   | `aws.roleArn` | Value of the ARN noted in an earlier step. |

#### Supported Aggregation Behavior

You can use AWS Kinesis connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stages.

### Add an HTTPS Connection through the Atlas Administration API

The Atlas Administration API provides an endpoint for adding a connection to a connection registry.

[Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection)

When you create an HTTPS connection through the Atlas Administration API, you work with two distinct sets of headers:

- **Request headers** are headers that establish the Atlas Administration API request itself. Send these with the `curl` command using the `--header` flag.

- **HTTPS connection headers** are optional runtime headers that Atlas Stream Processing sends to your external endpoint. Define these in the `headers` field of the connection body. If your endpoint requires authentication, include those credentials as connection headers.

**Example:**

The following example creates an HTTPS connection with both Atlas Administration API request headers and HTTPS connection headers:

```sh
curl --user "<publicApiKey>:<privateApiKey>" --digest \
  --header "Content-Type: application/json" \
  --header "Accept: application/vnd.atlas.2023-02-01+json" \
  --include \
  --data '{
    "name": "HTTPSConnection",
    "type": "Https",
    "url": "<apiBasePath>",
    "headers": {
      "Content-Type": "application/json",
      "Authorization": "Bearer <accessToken>"
    }
  }' \
  --request POST "https://cloud.mongodb.com/api/atlas/v2/groups/<projectID>/streams/<tenantName>/connections"
```

#### Supported Aggregation Behavior

You can use HTTPS connections only in the [$https](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#std-label-atlas-sp-agg-https) stage.

To learn how to use HTTPS connections with Atlas Stream Processing, see [`$https` Aggregation Stage (Stream Processing).](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#std-label-atlas-sp-agg-https)

### Add an S3 Connection through the Atlas Administration API

To add an S3 (Simple Storage Service) connection to your stream processing workspace through the Atlas Administration API, follow these steps:

1. Set up Unified AWS Access.

   Follow the procedure described in [Set Up Unified AWS Access.](https://docs.mongodb.com/atlas/security/set-up-unified-aws-access/?interface=atlas-admin-api)

   Ensure that your IAM policy grants the necessary permissions for your use case.

   - Writing to an S3 (Simple Storage Service) bucket with `$emit`

   - Writing to an [Apache Iceberg](https://iceberg.apache.org/docs/latest/) table with the standard `hadoop` catalog type

   - Writing to an [Apache Iceberg](https://iceberg.apache.org/docs/latest/) table with the AWS (Amazon Web Services) `glue` catalog type

   Use the following examples to guide your policy creation.

   ###### $emit

   ###### IAM policy for standard S3 $emit

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:ListBucket",
           "s3:GetObject",
           "s3:GetObjectVersion",
           "s3:GetBucketLocation"
         ],
         "Resource": [
           "arn:aws:s3:::bucketName",
           "arn:aws:s3:::bucketName/*"
         ]
       },
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
         ],
         "Resource": [
           "arn:aws:s3:::bucketName",
           "arn:aws:s3:::bucketName/*"
         ]
       }
     ]
   }
   ```

   ###### $iceberg

   ###### IAM policy for standard $iceberg output

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
           "s3:ListBucket",
           "s3:GetObject",
           "s3:GetObjectVersion",
           "s3:GetBucketLocation",
           "s3:AbortMultipartUpload",
           "s3:DeleteObject"
         ],
         "Resource": [
             "arn:aws:s3:::iceberg-asp-dev",
             "arn:aws:s3:::iceberg-asp-dev/*"
         ]
       }
     ]
   }
   ```

   ###### $iceberg with Glue

   ###### IAM policy for $iceberg with the Glue catalog type

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
             "glue:CreateDatabase",
             "glue:GetTable",
             "glue:CreateTable",
             "glue:GetDatabase",
             "glue:UpdateTable"
         ],
         "Resource": [
             "arn:aws:glue:us-east-1:1111111111:catalog",
             "arn:aws:glue:us-east-1:1111111111:database/asp-iceberg-db",
             "arn:aws:glue:us-east-1:1111111111:table/asp-iceberg-db/*"
         ]
       },
       {
         "Effect": "Allow",
         "Action": [
             "s3:PutObject",
             "s3:ListBucket",
             "s3:GetObject",
             "s3:GetObjectVersion",
             "s3:GetBucketLocation",
             "s3:AbortMultipartUpload",
             "s3:DeleteObject"
         ],
         "Resource": [
             "arn:aws:s3:::iceberg-asp-dev",
             "arn:aws:s3:::iceberg-asp-dev/*"
         ]
       }
     ]
   }
   ```

   Note the ARN value in `Statement.Principal.AWS` for later in this procedure.

2. In Atlas, go to the Users page.

   If it's not already displayed, select your desired organization from the  Organizations menu in the navigation bar.

   Click All Projects in the sidebar under the Identity & Access section, and select your desired project.

   Click Project Identity & Access in the sidebar under the Security section.

   The [Users](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2Faccess) page displays.

3. Create an API key.

   The [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) API endpoint requires digest authorization when creating an S3 (Simple Storage Service) connection. To support this, you must create an API Key.

   In the Project Access Manager, select the Applications tab, then click API Keys.

   Click Create API Key. Provide a short description for the key.

   In the Project Permissions dropdown menu, select both the Project Stream Processing Owner and Project Owner roles. Click Next.

   Save both the public and private keys for later in this procedure.

4. Create the S3 Connection.

   The Atlas Administration API provides an endpoint to [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection). You must send this request using digest authorization.

   In your HTTP request interface, enable digest authorization. For the `username`, provide the public key you generated previously. For the `password`, provide the private key you generated previously.

   For an AWS (Amazon Web Services) S3 (Simple Storage Service) connection, set the following key-value pairs:

   | Key | Value |
   | --- | --- |
   | `type` | `"S3"` |
   | `aws.roleArn` | Value of the ARN noted in an earlier step. |

#### Supported Aggregation Behavior

You can use AWS (Amazon Web Services) S3 (Simple Storage Service) connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add an Azure Blob Storage Connection through the Atlas Administration API

To add an Azure (Microsoft Azure) Blob Storage connection to your stream processing workspace through the Atlas Administration API, follow these steps:

1. Set up Azure Service Principal Access.

   Follow the procedure described in [Set Up and Manage Azure Service Principal Access.](https://www.mongodb.com/docs/atlas/security/set-up-azure-access.md#std-label-manage-azure-access)

   Grant your Service Principal the following permissions:

   - `Storage Blob Data Contributor` for your storage account

   - `Storage Blob Data Contributor` for your blob storage containers

   - `Reader` permissions for the Azure Subscription containing both the storage account and blob containers.

   Note the `Service Principal ID` you used for later in this procedure.

2. Configure a Service Account

   Create a [service account.](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-atlas-admin-api-access)

   [Grant](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-grant-api-access-project) this service account the following roles within your Atlas Stream Processing project:

   - [`Project Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Owner)

   - [`Project Stream Processing Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Stream-Processing-Owner)

3. Create the Azure (Microsoft Azure) Blob Storage connection.

   Make a request to the Atlas Administration API [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) endpoint according to the [procedure](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-example-api-request) described in the Service Accounts documentation.

   For values specific to an Azure (Microsoft Azure) Blob Storage connection, see the [field descriptions.](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection#operation-creategroupstreamconnection-body-application-vnd-atlas-2023-02-01-json-azureblobstorage-object)

#### Supported Aggregation Behavior

You can use Azure (Microsoft Azure) Blob Storage connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add a GCP Pub/Sub Connection through the Atlas Administration API

To add a Google Cloud Pub/Sub connection to your stream processing workspace through the Atlas Administration API, follow these steps:

1. Set up GCP Service Account Access.

   Follow the procedure described in [Set Up and Manage GCP Service Account Access.](https://www.mongodb.com/docs/atlas/security/set-up-gcp-access.md#std-label-manage-gcp-access)

   Note the `Service Account` ID for later in this procedure.

   Ensure that you grant the following roles at both the project and topic levels:

   - [Publisher Role](https://docs.cloud.google.com/pubsub/docs/access-control#pubsub.publisher)

   - [Viewer Role](https://docs.cloud.google.com/pubsub/docs/access-control#pubsub.viewer)

   For more information, see the [Pub/Sub documentation.](https://docs.cloud.google.com/pubsub/docs/access-control#console)

2. Configure an Atlas Service Account.

   Create a [service account.](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-atlas-admin-api-access)

   [Grant](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-grant-api-access-project) this service account the following roles within your Atlas Stream Processing project:

   - [`Project Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Owner)

   - [`Project Stream Processing Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Stream-Processing-Owner)

3. Create the Google Cloud Pub/Sub connection.

   Make a request to the Atlas Administration API [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) endpoint according to the [procedure](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-example-api-request) described in the Service Accounts documentation.

   For values specific to a GCP Pub/Sub connection, see the [field descriptions.](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection#operation-creategroupstreamconnection-body-application-vnd-atlas-2023-02-01-json-gcppubsub-object)

#### Supported Aggregation Behavior

You can use Google Cloud Pub/Sub connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add an AWS Lambda Connection through the Atlas Administration API

The Atlas Administration API provides the [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) endpoint for adding a connection to a connection registry.

**Note:**

This `curl` command uses a service account access token ([OAuth 2.0](https://oauth.net/2/)) to authenticate instead of API keys. To learn more, see [Get Started with the Atlas Administration API.](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-atlas-admin-api-access)

You can update the `roleArn` placeholder in the following example with the `arn` from your [AWS IAM configuration.](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns)

**Example:**

```bash
curl --header "Authorization: Bearer {ACCESS-TOKEN}" \
   --header "Content-Type: application/json" \
   --header "Accept: application/vnd.atlas.2023-02-01+json" \
   --include \
   --data '{"name": "TestAWSLambdaConnection","type": "AWSLambda","aws": {"roleArn": "arn:aws:iam::<aws_account>:role/<role_name>"}}' \
   --request POST "https://cloud.mongodb.com/api/atlas/v2/groups/<group_id>/streams/<tenant_name>/connections"
```

#### Supported Aggregation Behavior

You can use AWS (Amazon Web Services) Lambda connections only in the [$externalFunction](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#std-label-atlas-sp-agg-external-function) stage.

### Add a Schema Registry Connection through the Atlas Administration API

To add a [Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html) connection to your stream processing workspace through the Atlas Administration API, follow these steps:

1. Create a Schema Registry.

   Follow the procedure outlined in the [Confluent Cloud documentation.](https://docs.confluent.io/cloud/current/get-started/schema-registry.html)

2. Create a Confluent Cloud API Key.

   Follow the procedure outlined in the [Confluent Cloud documentation](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.html#add-an-api-key) to create a Confluent Cloud API Key for your Schema Registry.

3. Create the Schema Registry Connection.

   The Atlas Administration API provides an endpoint to [Create One Connection.](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection)

   For a Confluent Cloud Schema Registry connection, set all required key-value pairs. The following must have specific values:

   | Key | Value |
   | --- | --- |
   | `type` | `"SchemaRegistry"` |
   | `provider` | `"CONFLUENT"` |
   | `schemaRegistryUrls` | The URLs associated with your schema registry. |
   | `schemaRegistryAuthentication.type` | `"USER_INFO"` or `"SASL_INHERIT"`. If you use `"USER_INFO"`, you must also provide values for `schemaRegistryAuthentication.username` and `schemaRegistryAuthentication.password`. |
   | `schemaRegistryAuthentication.username` | The username associated with your Confluent Cloud API Key. |
   | `schemaRegistryAuthentication.password` | The password associated with your Confluent Cloud API Key. |

#### Supported Aggregation Behavior

You configure Schema Registry connections as a property of the [Apache Kafka](https://kafka.apache.org/) [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add a Kafka Connection through the Atlas UI

To add a Kafka connection to your stream processing workspace through the Atlas UI, follow these steps:

1. In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

2. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

3. Click + Add Connection.

4. Add a new connection.

   Select a Kafka connection.

   Provide a Connection Name. Each connection name must be unique within a stream processing workspace.  This setting is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   Select a Network Access type. Atlas Stream Processing supports Public IP or VPC Peering connections.

   ### Public IP

   Click the Public IP button. No further configuration is needed for this network access type.

   1. Specify an IP address for one or more [bootstrap servers](https://kafka.apache.org/documentation/#streamsconfigs_bootstrap.servers) for your [Apache Kafka](https://kafka.apache.org/) system.

   2. From the dropdown menu, select a Security Protocol
      Method. Atlas Stream Processing supports `SASL_PLAINTEXT`, `SASL_SSL`, and `SSL`.

      ### Tab

      `SASL_PLAINTEXT` is incompatible with VPC peering and Private Link connections. To use VPC peering or Private Link, you must select the `SASL_SSL` or the `SSL` method.

      From the dropdown menu, select a SASL Mechanism.

      Atlas Stream Processing supports:

      - `PLAIN`

      - `SCRAM-SHA-256`

      - `SCRAM-SHA-512`

      - `OAUTHBEARER`

      If you specify `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`, you must also provide the following information:

      1. A Username for authentication.

      2. A password for authentication.

      If you select `OAUTHBEARER`, you must also provide the following:

      1. Token endpoint URL - OAUTH issuer (IdP provider) token endpoint HTTP(S) URI used to retrieve the token.

      2. Client ID - Public identifier for the Kafka client.

      3. Client secret - Secret known only to the Kafka client and the authorization server.

      If you select `OAUTHBEARER`, you can also provide the following optional fields:

      1. Scope - Scope of the access request to the broker specified by the Kafka clients.

      2. SASL Extensions - Additional information to provide to the Kafka broker.

      Click + Add Connection.

   3. **(Optional)** If you plan to use this connection to create a stream processor with [failover processor](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover) functionality, configure the Failover region settings. You can either configure a unique Kafka broker for the failover region according to the preceding steps, or you can toggle the Copy Kafka broker settings from default region to use your existing broker configuration.

      IMPORTANT: Atlas Stream Processing supports failover processors using Kafka connections only as sinks with the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

   4. Click + Add Connection.

**Important:**

After adding an external connection such as an [Apache Kafka](https://kafka.apache.org/) cluster to your connection registry, you must add Atlas IP addresses to an access list for that external connection. To learn more, see [Allow Access to or from the Atlas Control Plane.](https://www.mongodb.com/docs/atlas/setup-cluster-security.md#std-label-atlas-add-inbound-ips)

#### Parameters Configuration

Each interface for creating a Kafka connection allows you to provide configuration parameters for your Kafka cluster. These configurations take the form of key-value pairs, and correspond to one of the following:

- A subset of [Producer](https://kafka.apache.org/documentation/#producerconfigs) configurations

- A subset of [Consumer](https://kafka.apache.org/documentation/#consumerconfigs) configurations.

Atlas Stream Processing passes only these parameters to your Kafka cluster. If you declare any parameters not explicitly allowed, Atlas Stream Processing ignores them.

| Interface | Configuration Mechanism |
| --- | --- |
| Atlas CLI | Provide configurations as a `.json` file specified with the `--file` flag. |
| Atlas Administration API | Provide configurations as a `.json` object specified in the `config` field. |
| Atlas UI | Provide configurations in the Configuration File field of the Add Connection page. |

Atlas Stream Processing supports the following configurations:

### Producer Parameters

- [acks](https://kafka.apache.org/documentation/#producerconfigs_acks)

- [batch.size](https://kafka.apache.org/documentation/#producerconfigs_batch.size)

- [client.dns.lookup](https://kafka.apache.org/documentation/#connectconfigs_client.dns.lookup)

- [client.id](https://kafka.apache.org/documentation/#consumerconfigs_client.id)

- [compression.type](https://kafka.apache.org/documentation/#producerconfigs_compression.type)

- [connections.max.idle.ms](https://kafka.apache.org/documentation/#brokerconfigs_connections.max.idle.ms)

- [delivery.timeout.ms](https://kafka.apache.org/documentation/#producerconfigs_delivery.timeout.ms)

- [enable.idempotence](https://kafka.apache.org/documentation/#producerconfigs_enable.idempotence)

- [linger.ms](https://kafka.apache.org/documentation/#producerconfigs_linger.ms)

- [max.in.flight.requests.per.connection](https://kafka.apache.org/documentation/#producerconfigs_max.in.flight.requests.per.connection)

- [message.max.bytes](https://kafka.apache.org/documentation/#brokerconfigs_message.max.bytes)

- [queue.buffering.max.messages](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md)

- [request.timeout.ms](https://kafka.apache.org/documentation/#producerconfigs_request.timeout.ms)

- [retries](https://kafka.apache.org/documentation/#producerconfigs_retries)

- [transactional.id](https://kafka.apache.org/documentation/#producerconfigs_transactional.id)

#### Supported Aggregation Behavior

You can use [Apache Kafka](https://kafka.apache.org/) connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stages.

### Add an Atlas Cluster Connection through the Atlas UI

To add an Atlas Cluster connection to your stream processing workspace through the Atlas UI, follow these steps:

1. **(Optional)** Enable cross-project connections.

   To create a cross-project Atlas connection, first enable this functionality. To learn more, see [Cross-Project Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections-cross-project)

2. In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

3. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

4. Click + Add Connection.

5. Add a new connection.

   To create a new connection to an Atlas [change stream:](https://www.mongodb.com/docs/manual/changeStreams/)

   Select an Atlas Database connection.

   Provide a Connection Name. Each connection name must be unique within an stream processing workspace. This is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   From the dropdown menu, select an Atlas
   Cluster. Atlas Stream Processing can connect only to dedicated-tier clusters. If you enable cross-project support, the available clusters appear grouped by project.

   **(Optional)** If you plan to use this connection to create a stream processor with [failover processor](https://www.mongodb.com/docs/atlas/atlas-stream-processing/architecture.md#std-label-atlas-sp-architecture-failover) functionality, configure the Failover region settings.

   Click + Add Connection.

#### Supported Aggregation Behavior

You can use Atlas cluster connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$merge](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-merge.md#std-label-atlas-sp-agg-merge) stages.

### Add a Kinesis Connection through the Atlas UI

**Important:**

Atlas Stream Processing currently supports using AWS (Amazon Web Services) Kinesis connections only as sinks.

To add a Kinesis connection to your stream processing workspace through Atlas UI, follow these steps:

1. Set up Unified AWS Access.

   Follow the procedure described in [Set Up Unified AWS Access.](https://docs.mongodb.com/atlas/security/set-up-unified-aws-access/?interface=atlas-ui)

   Ensure that you grant your IAM role the following permissions:

   - `SubscribeToShard`

   - `ListShards`

   - `DescribeStreamSummary`

   - `PutRecords`

   Note the ARN value in `Statement.Principal.AWS` for later in this procedure.

2. In Atlas, go to the Users page.

   If it's not already displayed, select your desired organization from the  Organizations menu in the navigation bar.

   Click All Projects in the sidebar under the Identity & Access section, and select your desired project.

   Click Project Identity & Access in the sidebar under the Security section.

   The [Users](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2Faccess) page displays.

3. Create an API key.

   The [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) API endpoint requires digest authorization when creating an S3 (Simple Storage Service) connection. To support this, you must create an API Key.

   In the Project Access Manager, select the Applications tab, then click API Keys.

   Click Create API Key. Provide a short description for the key.

   In the Project Permissions dropdown menu, select both the Project Stream Processing Owner and Project Owner roles. Click Next.

   Save both the public and private keys for later in this procedure.

   In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

4. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

5. Click + Add Connection.

6. Add a new connection.

   Select an AWS Kinesis connection.

   Provide a Connection Name. Each connection name must be unique within a stream processing workspace.  This is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   Under Network Access, select Public
   IP.

   From the AWS IAM Role ARN dropdown, select the ARN of the unified access role you authorized in a prior step.

   Click + Add Connection.

#### Supported Aggregation Behavior

You can use AWS Kinesis connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stages.

### Add an HTTPS Connection through the Atlas UI

To add an HTTPS connection to your stream processing workspace through Atlas UI, follow these steps:

1. Navigate to your stream processing workspace.

   Select the Stream Processing tab in the left sidebar menu in the Atlas UI.

2. Configure your connection.

   1. Click the Manage button for the stream processing workspace to which you want to register a connection.

   2. Select the Connection Registry tab.

   3. Click the + Add Connection button.

   4. Select HTTPS from the Connection Type dropdown menu.

   5. Name your new connection.

   6. Populate the URL input field with the URL your stream processor will make requests to.

   7. Add any required headers to be sent with the HTTPS request, such as the data type or authentication credentials required by the external API.

   8. Click the Create Connection button to create the connection.

#### Supported Aggregation Behavior

You can use HTTPS connections only in the [$https](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#std-label-atlas-sp-agg-https) stage.

### Add an S3 Connection through Atlas UI

To add an S3 (Simple Storage Service) connection to your stream processing workspace through Atlas UI, follow these steps:

1. Set up Unified AWS Access.

   Follow the procedure described in [Set Up Unified AWS Access.](https://docs.mongodb.com/atlas/security/set-up-unified-aws-access/?interface=atlas-ui)

   Ensure that your IAM policy grants the necessary permissions for your use case:

   - Writing to an S3 (Simple Storage Service) bucket with `$emit`

   - Writing to an [Apache Iceberg](https://iceberg.apache.org/docs/latest/) table with the standard `hadoop` catalog type

   - Writing to an [Apache Iceberg](https://iceberg.apache.org/docs/latest/) table with the AWS (Amazon Web Services) `glue` catalog type

   Use the following examples to guide your policy creation.

   ###### $emit

   ###### IAM policy for standard S3 $emit

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:ListBucket",
           "s3:GetObject",
           "s3:GetObjectVersion",
           "s3:GetBucketLocation"
         ],
         "Resource": [
           "arn:aws:s3:::bucketName",
           "arn:aws:s3:::bucketName/*"
         ]
       },
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
         ],
         "Resource": [
           "arn:aws:s3:::bucketName",
           "arn:aws:s3:::bucketName/*"
         ]
       }
     ]
   }
   ```

   ###### $iceberg

   ###### IAM policy for standard $iceberg output

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
           "s3:ListBucket",
           "s3:GetObject",
           "s3:GetObjectVersion",
           "s3:GetBucketLocation",
           "s3:AbortMultipartUpload",
           "s3:DeleteObject"
         ],
         "Resource": [
             "arn:aws:s3:::iceberg-asp-dev",
             "arn:aws:s3:::iceberg-asp-dev/*"
         ]
       }
     ]
   }
   ```

   ###### $iceberg with Glue

   ###### IAM policy for $iceberg with the Glue catalog type

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
             "glue:CreateDatabase",
             "glue:GetTable",
             "glue:CreateTable",
             "glue:GetDatabase",
             "glue:UpdateTable"
         ],
         "Resource": [
             "arn:aws:glue:us-east-1:1111111111:catalog",
             "arn:aws:glue:us-east-1:1111111111:database/asp-iceberg-db",
             "arn:aws:glue:us-east-1:1111111111:table/asp-iceberg-db/*"
         ]
       },
       {
         "Effect": "Allow",
         "Action": [
             "s3:PutObject",
             "s3:ListBucket",
             "s3:GetObject",
             "s3:GetObjectVersion",
             "s3:GetBucketLocation",
             "s3:AbortMultipartUpload",
             "s3:DeleteObject"
         ],
         "Resource": [
             "arn:aws:s3:::iceberg-asp-dev",
             "arn:aws:s3:::iceberg-asp-dev/*"
         ]
       }
     ]
   }
   ```

   Ensure that you grant your IAM role `ListAllMyBuckets` and `PutObject` permissions.

   Note the ARN value in `Statement.Principal.AWS` for later in this procedure.

2. In Atlas, go to the Users page.

   If it's not already displayed, select your desired organization from the  Organizations menu in the navigation bar.

   Click All Projects in the sidebar under the Identity & Access section, and select your desired project.

   Click Project Identity & Access in the sidebar under the Security section.

   The [Users](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2Faccess) page displays.

3. Create an API key.

   The [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) API endpoint requires digest authorization when creating an S3 (Simple Storage Service) connection. To support this, you must create an API Key.

   In the Project Access Manager, select the Applications tab, then click API Keys.

   Click Create API Key. Provide a short description for the key.

   In the Project Permissions dropdown menu, select both the Project Stream Processing Owner and Project Owner roles. Click Next.

   Save both the public and private keys for later in this procedure.

   In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

4. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

5. Click + Add Connection.

6. Add a new connection.

   Select an S3 connection.

   Provide a Connection Name. Each connection name must be unique within a stream processing workspace.  This is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   From the AWS IAM Role ARN dropdown, select the ARN of the unified access role you authorized in a prior step.

   Click + Add Connection.

#### Supported Aggregation Behavior

You can use AWS (Amazon Web Services) S3 (Simple Storage Service) connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add an Azure Blob Storage Connection through Atlas UI

To add an Azure (Microsoft Azure) Blob Storage connection to your stream processing workspace through Atlas UI, follow these steps:

1. Set up Atlas Stream Processing permissions.

   Ensure you have at least one of the following roles to manage connections:

   - [`Project Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Owner)

   - [`Project Stream Processing Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Stream-Processing-Owner)

2. Set up Azure Service Principal Access.

   Follow the procedure described in [Set Up and Manage Azure Service Principal Access.](https://www.mongodb.com/docs/atlas/security/set-up-azure-access.md#std-label-manage-azure-access)

   Ensure that you grant the following permissions:

   - `Storage Blob Data Contributor` for your storage account

   - `Storage Blob Data Contributor` for your blob storage containers

   - `Reader` permissions for the Azure Subscription containing both the storage account and blob containers.

   Note the `Service Principal ID` you used for later in this procedure.

3. In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

4. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

5. Click + Add Connection.

6. Add a new connection.

   Select an Azure Blob Storage connection.

   Provide a Connection Name. Each connection name must be unique within a stream processing workspace.  This is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   From the Azure Service Principal dropdown, select the service principal defined earlier in this procedure.

   Provide the Storage Account Name for the [Azure Storage Account](https://learn.microsoft.com/en-us/azure/storage/common/storage-account-overview) to which you want to connect.

   Click + Add Connection.

#### Supported Aggregation Behavior

You can use Azure (Microsoft Azure) Blob Storage connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add a GCP Pub/Sub Connection through the Atlas UI

To add a Google Cloud Pub/Sub connection to your stream processing workspace through the Atlas UI, follow these steps:

1. Set up GCP Service Account Access.

   Follow the procedure described in [Set Up and Manage GCP Service Account Access.](https://www.mongodb.com/docs/atlas/security/set-up-gcp-access.md#std-label-manage-gcp-access)

   Note the `Service Account` ID for later in this procedure.

   Ensure that you grant the following roles at both the project and topic levels:

   - [Publisher Role](https://docs.cloud.google.com/pubsub/docs/access-control#pubsub.publisher)

   - [Viewer Role](https://docs.cloud.google.com/pubsub/docs/access-control#pubsub.viewer)

   For more information, see the [Pub/Sub documentation.](https://docs.cloud.google.com/pubsub/docs/access-control#console)

2. In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

3. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

4. Click + Add Connection.

5. Add a new connection.

   Select a GCP Pub/Sub connection.

   Provide a Connection Name. Each connection name must be unique within a stream processing workspace.  This is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   From the GCP Service Account dropdown, select the service account defined earlier in this procedure.

   Click + Add Connection.

#### Supported Aggregation Behavior

You can use Google Cloud Pub/Sub connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add an AWS Lambda Connection through the Atlas UI

To add an AWS (Amazon Web Services) Lambda connection to your stream processing workspace through Atlas UI, follow these steps:

1. Navigate to your stream processing workspace.

   Select the Stream Processing tab in the left sidebar menu in the Atlas UI.

2. Configure your connection.

   1. Click the Manage button for your stream processing workspace to which you want to register a connection.

   2. Select the Connection Registry tab.

   3. Click the + Add Connection button.

   4. Select AWS Lambda from the Connection Type dropdown menu.

   5. Name your new connection.

   6. Select an AWS IAM Role ARN from the dropdown menu. To learn more about creating an AWS IAM Role ARN, see [Set Up Unified AWS Access.](https://www.mongodb.com/docs/atlas/security/set-up-unified-aws-access.md#std-label-set-up-unified-aws-access)

   7. Click + Add Connection to create the connection.

#### Supported Aggregation Behavior

You can use AWS (Amazon Web Services) Lambda connections only in the [$externalFunction](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#std-label-atlas-sp-agg-external-function) stage.

### Add a Schema Registry Connection through the Atlas UI

To add a [Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html) connection to your stream processing workspace through the Atlas UI, follow these steps:

1. Create a Schema Registry.

   Follow the procedure outlined in the [Confluent Cloud documentation.](https://docs.confluent.io/cloud/current/get-started/schema-registry.html)

2. Create a Confluent Cloud API Key.

   Follow the procedure outlined in the [Confluent Cloud documentation](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.html#add-an-api-key) to create a Confluent Cloud API Key for your Schema Registry.

3. In Atlas, go to the Stream Processing page for your project.

   If it's not already displayed, select the organization that contains your project from the  Organizations menu in the navigation bar.

   If it's not already displayed, select your project from the Projects menu in the navigation bar.

   In the sidebar, click Stream Processing under the Streaming Data heading.

   The [Stream Processing](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2FstreamProcessing) page displays.

4. Go to the Connection Registry.

   Locate the overview panel of the stream processing workspace you want to modify and click Manage.

   Select the Connection Registry tab.

5. Click + Add Connection.

6. Add a new connection.

   Select a Schema Registry connection.

   Provide a Connection Name. Each connection name must be unique within a stream processing workspace.  This is the name used to reference the connection in Atlas Stream Processing [aggregations.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/stream-aggregation-stages.md#std-label-atlas-sp-aggregation)

   From the Authentication Type dropdown, select either `USER_INFO` or `SASL_INHERIT`. If you select `SASL_INHERIT`, the connection inherits the same credentials as the Kafka connection with which you pair the Schema Registry, and you won't provide values for Username or Password.

   In the Username field, provide the username associated with your Confluent Cloud API Key.

   In the Password field, provide the password associated with your Confluent Cloud API Key.

   In the Schema Registry URL(s) field, provide any URLs associated with your Schema Registry.

   Click + Add Connection.

#### Supported Aggregation Behavior

You configure Schema Registry connections as a property of the [Apache Kafka](https://kafka.apache.org/) [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add a Kafka Connection through the Atlas CLI

To add a Kafka connection to your stream processing workspace through the Atlas CLI, follow these steps:

To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas streams connections create [connectionName] [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

**Important:**

After adding an external connection such as an [Apache Kafka](https://kafka.apache.org/) cluster to your connection registry, you must add Atlas IP addresses to an access list for that external connection. To learn more, see [Allow Access to or from the Atlas Control Plane.](https://www.mongodb.com/docs/atlas/setup-cluster-security.md#std-label-atlas-add-inbound-ips)

#### Parameters Configuration

Each interface for creating a Kafka connection allows you to provide configuration parameters for your Kafka cluster. These configurations take the form of key-value pairs, and correspond to one of the following:

- A subset of [Producer](https://kafka.apache.org/documentation/#producerconfigs) configurations

- A subset of [Consumer](https://kafka.apache.org/documentation/#consumerconfigs) configurations.

Atlas Stream Processing passes only these parameters to your Kafka cluster. If you declare any parameters not explicitly allowed, Atlas Stream Processing ignores them.

| Interface | Configuration Mechanism |
| --- | --- |
| Atlas CLI | Provide configurations as a `.json` file specified with the `--file` flag. |
| Atlas Administration API | Provide configurations as a `.json` object specified in the `config` field. |
| Atlas UI | Provide configurations in the Configuration File field of the Add Connection page. |

Atlas Stream Processing supports the following configurations:

### Producer Parameters

- [acks](https://kafka.apache.org/documentation/#producerconfigs_acks)

- [batch.size](https://kafka.apache.org/documentation/#producerconfigs_batch.size)

- [client.dns.lookup](https://kafka.apache.org/documentation/#connectconfigs_client.dns.lookup)

- [client.id](https://kafka.apache.org/documentation/#consumerconfigs_client.id)

- [compression.type](https://kafka.apache.org/documentation/#producerconfigs_compression.type)

- [connections.max.idle.ms](https://kafka.apache.org/documentation/#brokerconfigs_connections.max.idle.ms)

- [delivery.timeout.ms](https://kafka.apache.org/documentation/#producerconfigs_delivery.timeout.ms)

- [enable.idempotence](https://kafka.apache.org/documentation/#producerconfigs_enable.idempotence)

- [linger.ms](https://kafka.apache.org/documentation/#producerconfigs_linger.ms)

- [max.in.flight.requests.per.connection](https://kafka.apache.org/documentation/#producerconfigs_max.in.flight.requests.per.connection)

- [message.max.bytes](https://kafka.apache.org/documentation/#brokerconfigs_message.max.bytes)

- [queue.buffering.max.messages](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md)

- [request.timeout.ms](https://kafka.apache.org/documentation/#producerconfigs_request.timeout.ms)

- [retries](https://kafka.apache.org/documentation/#producerconfigs_retries)

- [transactional.id](https://kafka.apache.org/documentation/#producerconfigs_transactional.id)

#### Supported Aggregation Behavior

You can use [Apache Kafka](https://kafka.apache.org/) connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stages.

### Add an Atlas Cluster Connection through the Atlas CLI

IMPORTANT: To create a cross-project Atlas connection, first enable the functionality. To learn more, see [Cross-Project Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections-cross-project)

To add an Atlas Cluster connection to your stream processing workspace through the Atlas CLI, follow these steps:

To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

```sh

atlas streams connections create [connectionName] [options]

```

To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

When you create a stream processing workspace using the Atlas CLI, you must provide a `.json` configuration file to define the new connection's parameters. The format of this file depends on the type of connection you define.

For an Atlas database connection, provide a configuration file with the following syntax:

```json
{
   "type": "Cluster",
   "clusterName": "<clusterName>",
   "dbRoleToExecute": {
     "role": "<roleName>",
     "type": "<roleType>"
   }
}
```

#### Supported Aggregation Behavior

You can use Atlas cluster connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$merge](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-merge.md#std-label-atlas-sp-agg-merge) stages.

### Add a Kinesis Connection through the Atlas CLI

To add an AWS Kinesis connection to your stream processing workspace through the Atlas CLI, follow these steps:

1. Set up Unified AWS Access.

   Follow the procedure described in [Set Up Unified AWS Access.](https://docs.mongodb.com/atlas/security/set-up-unified-aws-access/?interface=atlas-admin-api)

   Ensure that you grant your IAM role the following permissions:

   - `SubscribeToShard`

   - `ListShards`

   - `DescribeStreamSummary`

   - `PutRecords`

   Note the ARN value in `Statement.Principal.AWS` for later in this procedure.

2. In Atlas, go to the Users page.

   If it's not already displayed, select your desired organization from the  Organizations menu in the navigation bar.

   Click All Projects in the sidebar under the Identity & Access section, and select your desired project.

   Click Project Identity & Access in the sidebar under the Security section.

   The [Users](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2Faccess) page displays.

3. Create an API key.

   The [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) API endpoint requires digest authorization when creating an S3 (Simple Storage Service) connection. To support this, you must create an API Key.

   In the Project Access Manager, select the Applications tab, then click API Keys.

   Click Create API Key. Provide a short description for the key.

   In the Project Permissions dropdown menu, select both the Project Stream Processing Owner and Project Owner roles. Click Next.

   Save both the public and private keys for later in this procedure.

4. Create the Kinesis connection.

   To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

   ```sh

   atlas streams connections create [connectionName] [options]

   ```

   To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

   In your configuration file, set the following key-value pairs:

   | Key | Value |
   | --- | --- |
   | `type` | `"AWSKinesisDataStreams"` |
   | `aws.roleArn` | Value of the ARN noted in an earlier step. |

#### Supported Aggregation Behavior

You can use AWS Kinesis connections in the [$source](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#std-label-atlas-sp-agg-source) and [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stages.

### Add an HTTPS Connection through the Atlas CLI

To add an HTTPS connection to your stream processing workspace through Atlas CLI, follow these steps:

1. Create a config.json file.

   ```javascript
   {
      "name": "ExampleHTTPSConnection",
      "type": "Https",
   }
   ```

2. Create the HTTPS connection.

   Run the following Atlas CLI command to create the connection:

   ```sh
   atlas streams connections create -i <instance-name> -f </path/to/config.json>
   ```

#### Supported Aggregation Behavior

You can use HTTPS connections only in the [$https](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-https.md#std-label-atlas-sp-agg-https) stage.

### Add an S3 Connection through Atlas CLI

To add an S3 (Simple Storage Service) connection to your stream processing workspace through the Atlas CLI, follow these steps:

1. Set up Unified AWS Access.

   Follow the procedure described in [Set Up Unified AWS Access.](https://docs.mongodb.com/atlas/security/set-up-unified-aws-access/?interface=atlas-admin-api)

   Ensure that your IAM policy grants the necessary permissions for your use case.

   - Writing to an S3 (Simple Storage Service) bucket with `$emit`

   - Writing to an [Apache Iceberg](https://iceberg.apache.org/docs/latest/) table with the standard `hadoop` catalog type

   - Writing to an [Apache Iceberg](https://iceberg.apache.org/docs/latest/) table with the AWS (Amazon Web Services) `glue` catalog type

   Use the following examples to guide your policy creation.

   ###### $emit

   ###### IAM policy for standard S3 $emit

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:ListBucket",
           "s3:GetObject",
           "s3:GetObjectVersion",
           "s3:GetBucketLocation"
         ],
         "Resource": [
           "arn:aws:s3:::bucketName",
           "arn:aws:s3:::bucketName/*"
         ]
       },
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
         ],
         "Resource": [
           "arn:aws:s3:::bucketName",
           "arn:aws:s3:::bucketName/*"
         ]
       }
     ]
   }
   ```

   ###### $iceberg

   ###### IAM policy for standard $iceberg output

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "s3:PutObject",
           "s3:ListBucket",
           "s3:GetObject",
           "s3:GetObjectVersion",
           "s3:GetBucketLocation",
           "s3:AbortMultipartUpload",
           "s3:DeleteObject"
         ],
         "Resource": [
             "arn:aws:s3:::iceberg-asp-dev",
             "arn:aws:s3:::iceberg-asp-dev/*"
         ]
       }
     ]
   }
   ```

   ###### $iceberg with Glue

   ###### IAM policy for $iceberg with the Glue catalog type

   ```json
   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
             "glue:CreateDatabase",
             "glue:GetTable",
             "glue:CreateTable",
             "glue:GetDatabase",
             "glue:UpdateTable"
         ],
         "Resource": [
             "arn:aws:glue:us-east-1:1111111111:catalog",
             "arn:aws:glue:us-east-1:1111111111:database/asp-iceberg-db",
             "arn:aws:glue:us-east-1:1111111111:table/asp-iceberg-db/*"
         ]
       },
       {
         "Effect": "Allow",
         "Action": [
             "s3:PutObject",
             "s3:ListBucket",
             "s3:GetObject",
             "s3:GetObjectVersion",
             "s3:GetBucketLocation",
             "s3:AbortMultipartUpload",
             "s3:DeleteObject"
         ],
         "Resource": [
             "arn:aws:s3:::iceberg-asp-dev",
             "arn:aws:s3:::iceberg-asp-dev/*"
         ]
       }
     ]
   }
   ```

   Note the ARN value in `Statement.Principal.AWS` for later in this procedure.

2. In Atlas, go to the Users page.

   If it's not already displayed, select your desired organization from the  Organizations menu in the navigation bar.

   Click All Projects in the sidebar under the Identity & Access section, and select your desired project.

   Click Project Identity & Access in the sidebar under the Security section.

   The [Users](https://cloud.mongodb.com/go?l=https%3A%2F%2Fcloud.mongodb.com%2Fv2%2F%3Cproject%3E%23%2Faccess) page displays.

3. Create an API key.

   The [Create One Connection](https://www.mongodb.com/docs/api/doc/atlas-admin-api-v2/operation/operation-creategroupstreamconnection) API endpoint requires digest authorization when creating an S3 (Simple Storage Service) connection. To support this, you must create an API Key.

   In the Project Access Manager, select the Applications tab, then click API Keys.

   Click Create API Key. Provide a short description for the key.

   In the Project Permissions dropdown menu, select both the Project Stream Processing Owner and Project Owner roles. Click Next.

   Save both the public and private keys for later in this procedure.

4. Create the S3 connection.

   To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

   ```sh

   atlas streams connections create [connectionName] [options]

   ```

   To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

   In your configuration file, set the following key-value pairs:

   | Key | Value |
   | --- | --- |
   | `type` | `"S3"` |
   | `aws.roleArn` | Value of the ARN noted in an earlier step. |

#### Supported Aggregation Behavior

You can use AWS (Amazon Web Services) S3 (Simple Storage Service) connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add an Azure Blob Storage Connection through the Atlas CLI

To add an Azure (Microsoft Azure) Blob Storage connection to your stream processing workspace through the Atlas CLI, follow these steps:

1. Set up Azure Service Principal Access.

   Follow the procedure described in [Set Up and Manage Azure Service Principal Access.](https://www.mongodb.com/docs/atlas/security/set-up-azure-access.md#std-label-manage-azure-access)

   Ensure that you grant the following permissions:

   - `Storage Blob Data Contributor` for your storage account

   - `Storage Blob Data Contributor` for your blob storage containers

   - `Reader` permissions for the Azure Subscription containing both the storage account and blob containers.

   Note the `Service Principal ID` you used for later in this procedure.

2. Configure a Service Account.

   Create a [service account.](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-atlas-admin-api-access)

   [Grant](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-grant-api-access-project) this service account the following roles within your Atlas Stream Processing project:

   - [`Project Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Owner)

   - [`Project Stream Processing Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Stream-Processing-Owner)

3. Create the Azure (Microsoft Azure) Blob Storage connection.

   To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

   ```sh

   atlas streams connections create [connectionName] [options]

   ```

   To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

   In your configuration file, set the following key-value pairs:

   | Key | Value |
   | --- | --- |
   | `type` | `"AzureBlobStorage"` |
   | `servicePrincipalId` | Value of the `Service Principal ID` noted earlier. |
   | `storageAccountName` | Name of the [Azure Storage Account](https://learn.microsoft.com/en-us/azure/storage/common/storage-account-overview) to which you want to connect. |

#### Supported Aggregation Behavior

You can use Azure (Microsoft Azure) Blob Storage connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add a GCP Pub/Sub Connection through the Atlas CLI

To add a Google Cloud Pub/Sub connection to your stream processing workspace through the Atlas CLI, follow these steps:

1. Set up GCP Service Account Access.

   Follow the procedure described in [Set Up and Manage GCP Service Account Access.](https://www.mongodb.com/docs/atlas/security/set-up-gcp-access.md#std-label-manage-gcp-access)

   Note the `Service Account` ID for later in this procedure.

   Ensure that you grant the following roles at both the project and topic levels:

   - [Publisher Role](https://docs.cloud.google.com/pubsub/docs/access-control#pubsub.publisher)

   - [Viewer Role](https://docs.cloud.google.com/pubsub/docs/access-control#pubsub.viewer)

   For more information, see the [Pub/Sub documentation.](https://docs.cloud.google.com/pubsub/docs/access-control#console)

2. Configure a Service Account.

   Create a [service account.](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-atlas-admin-api-access)

   [Grant](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-grant-api-access-project) this service account the following roles within your Atlas Stream Processing project:

   - [`Project Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Owner)

   - [`Project Stream Processing Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Stream-Processing-Owner)

3. Create the Google Cloud Pub/Sub Connection

   To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

   ```sh

   atlas streams connections create [connectionName] [options]

   ```

   To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

   In your configuration file, set the following key-value pairs:

   | Key | Value |
   | --- | --- |
   | `type` | `"GCPPubSub"` |
   | `gcp.serviceAccountId` | Value of the `Service Account` noted earlier. |

#### Supported Aggregation Behavior

You can use Google Cloud Pub/Sub connections only in the [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

### Add an AWS Lambda Connection through the Atlas CLI

To add an AWS (Amazon Web Services) Lambda connection to your stream processing workspace through the Atlas CLI, follow these steps:

1. Create a `config.json` file.

   ```javascript
   {
      "name": "ExampleAWSLambdaConnection",
      "type": "AWSLambda",
      "aws": {
         "roleArn": "<your-aws-role-arn>"
      }
   }
   ```

2. Create the AWS (Amazon Web Services) Lambda connection.

   Run the following Atlas CLI command to create the connection:

   ```sh
   atlas streams connections create -i <instance-name> -f </path/to/config.json>
   ```

#### Supported Aggregation Behavior

You can use AWS (Amazon Web Services) Lambda connections only in the [$externalFunction](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#std-label-atlas-sp-agg-external-function) stage.

### Add a Schema Registry Connection through the Atlas CLI

To add a [Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html) connection to your stream processing workspace through the Atlas CLI, follow these steps:

1. Create a Schema Registry.

   Follow the procedure outlined in the [Confluent Cloud documentation.](https://docs.confluent.io/cloud/current/get-started/schema-registry.html)

2. Create a Confluent Cloud API Key.

   Follow the procedure outlined in the [Confluent Cloud documentation](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.html#add-an-api-key) to create a Confluent Cloud API Key for your Schema Registry.

3. Create the Schema Registry connection.

   To create one connection for the stream processing workspace you specify using the Atlas CLI, run the following command:

   ```sh

   atlas streams connections create [connectionName] [options]

   ```

   To learn more about the command syntax and parameters, see the Atlas CLI documentation for [atlas streams connections create.](https://www.mongodb.com/docs/atlas/cli/current/command/atlas-streams-connections-create/)

   In your configuration file, set all required key-value pairs. The following must have specific values:

   | Key | Value |
   | --- | --- |
   | `type` | `"SchemaRegistry"` |
   | `provider` | `"CONFLUENT"` |
   | `schemaRegistryUrls` | The URLs associated with your schema registry. |
   | `schemaRegistryAuthentication.type` | `"USER_INFO"` or `"SASL_INHERIT"`. If you use `"USER_INFO"`, you must also provide values for `schemaRegistryAuthentication.username` and `schemaRegistryAuthentication.password`. |
   | `schemaRegistryAuthentication.username` | The username associated with your Confluent Cloud API Key. |
   | `schemaRegistryAuthentication.password` | The password associated with your Confluent Cloud API Key. |

#### Supported Aggregation Behavior

You configure Schema Registry connections as a property of the [Apache Kafka](https://kafka.apache.org/) [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit) stage.

To learn how to configure Atlas Stream Processing connections with advanced security features, see the following pages:

- [VPC Peering Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-vpc-peering-connections.md#std-label-atlas-sp-manage-vpc-peering)

- [Apache Kafka Private Link Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/kafka-private-connection.md#std-label-atlas-sp-pl-kafka-add)

- [AWS S3 Private Link Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/s3-private-connection.md#std-label-atlas-sp-pl-s3-add)

- [AWS Kinesis Private Link Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/kinesis-private-connection.md#std-label-atlas-sp-pl-kinesis-add)

- [Google Cloud Pub/Sub Private Service Connect Connections](https://www.mongodb.com/docs/atlas/atlas-stream-processing/gcp-pubsub-private-connection.md#std-label-atlas-sp-pl-gcp-pubsub-add)