Setting the file. One moment.
Chapter 01 · MongoDB Atlas Stream Processing
Subchapter 1.1
references/connection-configs.mdMarkdown13 KBView on GitHub
Official examples repo: https://github.com/mongodb/ASP_example — check quickstarts, example processors, and Terraform examples. Start with quickstarts.
Know what each connection type can do before creating pipelines:
| Connection Type | As Source ($source) | As Sink ($merge / $emit) | Mid-Pipeline | Notes |
|---|---|---|---|---|
| Cluster | ✅ Change streams | ✅ $merge to collections | ✅ $lookup | Change streams monitor insert/update/delete/replace operations |
| Kafka | ✅ Topic consumer | ✅ $emit to topics | ❌ | Source MUST include topic field |
| Sample Stream | ✅ Sample data | ❌ Not valid | ❌ | Testing/demo only |
| S3 | ❌ Not valid | ✅ $emit to buckets | ❌ | Sink only - use path, format, compression |
| Https | ❌ Not valid | ✅ $https as sink | ✅ $https enrichment | Can be used mid-pipeline for enrichment OR as final sink stage |
| AWSLambda | ❌ Not valid | ✅ $externalFunction (async only) | ✅ $externalFunction (sync or async) | Sink: execution: "async" required. Mid-pipeline: execution: "sync" or "async" |
| AWS Kinesis | ✅ Stream consumer | ✅ $emit to streams | ❌ | Similar to Kafka pattern |
| SchemaRegistry | ❌ Not valid | ❌ Not valid | ✅ Schema resolution | Metadata only - used by Kafka connections for Avro schemas |
Common connection usage mistakes to avoid:
$source → HTTPS is for enrichment or sink only$externalFunction as sink with execution: "sync" → Must use execution: "async" for sink stage$merge with Kafka → Use $emit for Kafka sinks$externalFunction execution modes:
execution: "sync" (blocks until Lambda returns) or execution: "async" (non-blocking)execution: "async" onlyCRITICAL: Connection names should clearly indicate their actual targets to avoid confusion and prevent writing data to wrong destinations.
Match the actual target name:
cluster-restore-test or ClusterRestoreTestatlas-cluster or AtlasClusterUse descriptive names with context:
prod-kafka-orders (indicates environment + service + purpose)dev-atlas-main (indicates environment + service + designation)staging-s3-exports (indicates environment + service + purpose)❌ Generic names that don’t match targets:
❌ Reusing names across workspaces without context:
❌ Names that don’t indicate connection type:
Before creating processors, always inspect your connections to verify they point where you expect:
1. atlas-streams-discover → action: "list-connections"
2. atlas-streams-discover → action: "inspect-connection" for each
3. Verify connection name matches actual target (clusterName, bootstrapServers, url, etc.)
4. If mismatch exists, consider renaming or warn the userSee development-workflow.md “Pre-Deployment Connection Validation” section for the complete validation procedure.
$https enrichment ONLY — they are NOT valid as $source data sourcesconnectionType values: Kafka, Cluster, S3, Https, AWSKinesisDataStreams, AWSLambda, SchemaRegistry, SampleFor S3, Kinesis, and Lambda connections:
AWS connections (S3, Kinesis, Lambda) require that the IAM role ARN be registered in the Atlas project via Cloud Provider Access before creating the connection. This is a prerequisite — the connection creation will fail without it.
Always mention this prerequisite in your response when the user wants to create AWS connections, even if the user says connections already exist. Confirm with language like:
Security best practice: Use a dedicated IAM role per processor (or group of related processors) with least-privilege permissions scoped only to the specific S3 buckets, Kinesis streams, or Lambda functions that processor needs. Avoid sharing broad-access roles across unrelated processors.
The region field for workspace creation uses Atlas-specific names that differ by cloud provider. Using the wrong format returns a cryptic dataProcessRegion error.
| Provider | Cloud Region | Streams region Value |
|---|---|---|
| AWS | us-east-1 | VIRGINIA_USA |
| AWS | us-east-2 | OHIO_USA |
| AWS | us-west-2 | OREGON_USA |
| AWS | ca-central-1 | MONTREAL_CAN |
| AWS | sa-east-1 | SAOPAULO_BRA |
| AWS | eu-west-1 | DUBLIN_IRL |
| AWS | ap-southeast-1 | SINGAPORE_SGP |
| AWS | ap-south-1 | MUMBAI_IND |
| AWS | ap-northeast-1 | TOKYO_JPN |
| GCP | us-central1 | US_CENTRAL1 |
| GCP | europe-west1 | EUROPE_WEST1 |
| GCP | us-east4 | US_EAST4 |
| Azure | eastus | eastus |
| Azure | eastus2 | eastus2 |
| Azure | westus | westus |
| Azure | westeurope | westeurope |
This is a partial list. If unsure, inspect an existing workspace with atlas-streams-discover → inspect-workspace and check dataProcessRegion.region.
Elicitation: When required fields are missing, the build tool auto-prompts for them via an interactive form (MCP elicitation protocol). Do NOT manually ask the user for passwords or bootstrap servers — let the tool collect them.
Auto-normalization:
bootstrapServers passed as array → auto-converted to comma-separated stringschemaRegistryUrls passed as string → auto-wrapped in arraydbRoleToExecute → auto-defaults to {role: "readWriteAnyDatabase", type: "BUILT_IN"} if omitted{
"bootstrapServers": "broker1:9092,broker2:9092",
"authentication": {
"mechanism": "SCRAM-256",
"username": "my-user",
"password": "my-password"
},
"security": {
"protocol": "SASL_SSL"
}
}Important: bootstrapServers is a comma-separated string, not an array.
All fields above are required. The tool will prompt the user for username/password via elicitation if not provided.
Authentication mechanisms: PLAIN, SCRAM-256, SCRAM-512, OAUTHBEARER
Security protocols: SASL_SSL, SASL_PLAINTEXT, SSL
For Confluent Cloud, use mechanism: "PLAIN" with your API key as username and API secret as password.
Kafka supports both PrivateLink and VPC Peering for private networking. See the PrivateLink Reference section below for all supported vendors and providers.
VPC Peering:
SASL_SSL security protocolatlas-streams-manage with accept-peering action to complete the peering setupImportant: Networking cannot be modified after connection creation. To add or change PrivateLink/VPC peering on an existing Kafka connection, you must delete it and recreate it with the networking config.
Use atlas-streams-discover → action: "get-networking" to list available PrivateLink endpoints and VPC peering connections.
{
"clusterName": "my-atlas-cluster",
"dbRoleToExecute": {
"role": "readWriteAnyDatabase",
"type": "BUILT_IN"
}
}clusterName is required — must be a cluster in the same project (use atlas-list-clusters to verify).
dbRoleToExecute defaults to {role: "readWriteAnyDatabase", type: "BUILT_IN"} if not provided.
Optional: clusterGroupId (if cluster is in a different project — requires cross-project access to be enabled at the org level).
{
"aws": {
"roleArn": "arn:aws:iam::123456789:role/streams-s3-role",
"testBucket": "my-test-bucket"
}
}Prerequisite: The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.
Required IAM policy permissions: s3:ListBucket, s3:GetObject, s3:PutObject.
{
"url": "https://api.example.com/webhook",
"headers": {
"Authorization": "Bearer token123"
}
}IMPORTANT: HTTPS connections are for $https enrichment stages ONLY. They are NOT valid data sources — do not use them in $source.
Store all API authentication in the connection config headers, not in the processor pipeline.
API Key:
{"url": "https://api.example.com", "headers": {"X-API-Key": "your-api-key"}}Bearer Token:
{"url": "https://api.example.com", "headers": {"Authorization": "Bearer your-token"}}Basic Auth:
{"url": "https://api.example.com", "headers": {"Authorization": "Basic base64-encoded-credentials"}}OAuth 2.0 (pre-obtained token):
{"url": "https://api.example.com", "headers": {"Authorization": "Bearer oauth-access-token"}}{
"aws": {
"roleArn": "arn:aws:iam::123456789:role/streams-kinesis-role"
}
}Prerequisite: The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.
Required IAM policy permissions: kinesis:ListShards, kinesis:SubscribeToShard, kinesis:PutRecords, kinesis:DescribeStreamSummary.
{
"aws": {
"roleArn": "arn:aws:iam::123456789:role/streams-lambda-role"
}
}Prerequisite: The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.
{
"connectionType": "SchemaRegistry",
"connectionConfig": {
"schemaRegistryUrls": ["https://schema-registry.example.com"],
"schemaRegistryAuthentication": {
"type": "USER_INFO",
"username": "...",
"password": "..."
}
}
}connectionType MUST be "SchemaRegistry" (not "Kafka" or "Https")schemaRegistryUrls is an array (not a string). The tool auto-wraps a string into an array if needed.schemaRegistryAuthentication.type: "USER_INFO" (explicit credentials) or "SASL_INHERIT" (inherit from Kafka connection)No connectionConfig required. Provides built-in test data. Useful for development and testing without external infrastructure.
Available sample formats: sample_stream_solar (default, auto-created when includeSampleData: true on workspace), samplestock, sampleweather, sampleiot, samplelog, samplecommerce.
PrivateLink is supported for Kafka, S3, Kinesis, and Azure EventHub connections. Create a project-level PrivateLink first, then reference it in the connection’s networking.access config.
Step 1: Create project-level PrivateLink via atlas-streams-build resource=’privatelink’:
| Provider | Vendor | Required privateLinkConfig fields |
|---|---|---|
| AWS | CONFLUENT | provider, vendor, dnsDomain, dnsSubDomain (array, [] if none) |
| AWS | MSK | provider, vendor, arn |
| AWS | S3 | provider, vendor, region, serviceEndpointId (com.amazonaws.<region>.s3) |
| AWS | KINESIS | provider, vendor, region, serviceEndpointId |
| AZURE | EVENTHUB | provider, vendor, dnsDomain, serviceEndpointId |
| AZURE | CONFLUENT | provider, vendor, dnsDomain |
| GCP | CONFLUENT | provider, vendor, gcpServiceAttachmentUris |
Step 2: Reference in connection networking config:
{
"networking": {
"access": {
"type": "PRIVATE_LINK",
"connectionId": "<PrivateLink _id from Step 1>"
}
}
}Use atlas-streams-discover action=’get-networking’ to find the PrivateLink _id.
Note: Networking config cannot be modified after connection creation — delete and recreate to change.