This guide describes best practices for managing Atlas Stream Processing, including specific procedures for essential workflows.
Networking
Each stream processor relies on connections to data sources and sinks. Atlas Stream Processing encrypts all data in transit using TLS/SSL, but basic connections still pass data through the public internet. To achieve optimal performance and security with Atlas Stream Processing, consider the practices described in this section.
Peering and Route Tables
To ensure secure communication between Atlas Stream Processing and private external systems such as self-managed Apache Kafka clusters or private APIs, use VPC or VNet peering connections. These connections shield your data from exposure to the public internet. Note that in addition to establishing the peering connection, you must explicitly configure route tables for your application's VPC to direct traffic to your Atlas VPC CIDR block. To learn more about how to configure your route table, consult the VPC documentation for your external provider.
Atlas Cross-Project Connections
To route streaming data between different Atlas projects within the same organization, Atlas Stream Processing supports Cross Project Connections. Cross Project Connections transmit data without exposing clusters to the private internet, which ensures privacy without the need to manually configure VPC peering.
Passwordless AWS Authentication
Atlas Stream Processing supports the following AWS integrations:
S3
Kinesis Data Streams
Lambda (as
$externalFunction)
Each of these integrations is compatible with Unified AWS Access. Configuring Unified Access eliminates the need to store static access IDs or Secret credentials in MongoDB, and is the recommended authentication model for all AWS integrations.
Kafka OIDC
Atlas Stream Processing supports authenticating to Apache Kafka brokers using OIDC with JSON Web Tokens. However, this feature currently supports only Identity Providers accessible over public networks. This feature isn't supported for Identity Providers within a customer's VPC.
Fault Tolerance
Atlas Stream Processing provides various mechanisms to ensure reliable, robust service in common failure scenarios, including checkpoints, dead letter queues, and failover processors.
Atlas Stream Processing monitors stream processor health using a 30-second heartbeat. If a stream processor fails to send a heartbeat for more than 30 seconds, Atlas Stream Processing automatically restarts the processor.
Atlas Stream Processing automatically manages topology changes for sharded integrations like AWS Kinesis.
Dead Letter Queues
You can configure Atlas Stream Processing to route unprocessed data to a dead letter queue (DLQ). You can then view unprocessed records in the DLQ to remediate processing failures.
When you examine your DLQ, look for the following common processing errors:
Payload Parsing
The incoming record is malformed and fails to parse. This often results from invalid JSON.
Expression Evaluation
An expression in the processing pipeline fails to evaluate properly.
Example
Your pipeline includes the
$emit.config.tombstoneWhenconfiguration field to delete Kafka records and a given document is missing the field to be evaluated.Size Limit
The outgoing record exceeds the size limit of your target sink, such as
1 MBfor AWS Kinesis.Key Generation
The processor encounters an error when generating a partition key for your sink.
Late Data
Your processor uses windows and a given document arrived after the configured allowed lateness period expires.
After you analyze the causes of processing failures, modify either your upstream data sources to correct the data before it enters your Atlas Stream Processing pipeline, or modify your pipeline to handle edge cases.
Consider the following remediation options for records in the DLQ after using them to guide corrections to your Atlas Stream Processing configuration:
Discard
If a record is highly malformed or irrelevant—as in the case of very late data in highly time-sensitive systems—delete it from the collection.
Manual Correction
For a record with minor JSON errors or one that's mission-critical, manually adjust the fields and insert them directly into your target sink.
Automated Reprocessing:
For large volumes of recoverable, relevant data, create a secondary script or an additional stream processor that reads from the DLQ and applies a transformation against known failure conditions before routing the cleaned data to the target sink.
Failover
Atlas Stream Processing provides failover processors to prevent service outage in the event of region-wide system failure.
Principles of Failover Configuration
The following principles define a robust failover configuration to protect your Atlas Stream Processing workloads.
Deploy Atlas clusters in multiple regions.
To ensure continued operation of your stream processors in a regional failover event, any Atlas cluster to which your stream processor connects must be a multi-region cluster. This principle ensures your source or sink connections remain active when the primary region is down.
Create Atlas Stream Processing workspaces for each region.
After deploying your multi-region clusters, create a workspace for each region. This principle minimizes the latency of connections between your stream processors and their supporting Atlas databases.
Populate the Connection Registry with backup connections.
Configure connections for each source and sink in each region's workspace. Ensure that you configure connections not only for basic connections, but for VPC and private endpoint connections.
Configure connection-specific disaster recovery
To ensure continuity of service for a stream processor, any external providers on which it relies must also have failover enabled. Ensure that each external provider source or sink is properly configured according to its documentation.
Example
If your stream processor uses a Kafka cluster as a source or sink, configure the cluster for consumer offset preservation to enable Atlas Stream Processing to resume processing from a consumer group offset.
Perform routine dry runs.
Verify your failover configuration by periodically performing failover dry runs. These dry runs serve both to ensure your configuration connects to networks and loads checkpoints correctly and to train the failover workflow so that it is easier to perform at need. Consider establishing a regular cadence of failover dry runs, and perform a dry run every time you alter your failover configuration.
Setting Up Stream Processor Failover
After ensuring best practices are in place, create a stream processor and verify your failover configuration.
Atlas Stream Processing supports automated regional failover only for processors with the following configurations:
Atlas source and sink
Atlas source and Apache Kafka sink
For more information about automated failover, see Failover Processor.
However, you can configure an Atlas Stream Processing architecture that supports manual failover for any combination of source and sink connection.
Initiate Forced Failover
You can initiate a forced failover at either the workspace or stream processor level.
To initiate stream processing workspace failover, follow the procedure described in Trigger stream processing workspace Failover.
To initiate individual stream processor failover, follow the procedure described in Initiate Failover for One Stream Processor.
Responding to a Failover Event
In case of a regional service outage event:
Verify the service outage.
When you notice symptoms of a service outage, review your preferred status report to verify the event. You can view this information on the MongoDB Cloud Status page or your cloud provider's status page. You can configure Atlas Stream Processing Alerts. Note the timestamp of the failure event.
Start failover processors.
Follow the procedure described in Initiate Failover for
One Stream Processor. To ensure service resumes from the last consumed stream event, pass the timestamp you noted previously as the startAtOperationTime parameter.
Processors using Apache Kafka sources resume according to consumer group offsets.
(Optional) Automate primary processor suppression.
When your primary region comes back online, your primary processors will restart automatically, resulting in duplicate processing. Create and run a script to continuously check the status of the primary region and stop all processors in that region when it comes back online.
Verify that the service resumes.
After completing failover, periodically check the status of your workspace's primary region. You can view this information on the MongoDB Cloud Status page or your cloud provider's status page. You can also configure Atlas Stream Processing Alerts. If you use automatic primary processor suppression, you can configure the script to return notifications when it detects primary processors automatically restarting.
Restart primary processors.
Restart the stream processors in your primary region. To ensure service resumes from the last consumed stream event, pass the timestamp you noted previously as the startAtOperationTime parameter.
Processors using Apache Kafka sources resume according to consumer group offsets.
Failover Behavior
Consider the following when you configure your systems for regional failover or triggering manual failover:
When regional failover occurs, Atlas Stream Processing promotes the inactive failove processors in your failover region to active processors. These processors resume stream processing from the last available checkpoint.
Atlas Stream Processing checkpoints are guaranteed to be no more than 10 minutes old. However, failover processors reprocess any data that your primary processor processed since the last checkpoint on initiation. Failover guarantees at-least-once processing semantics.
If a failover processor fails to start during a regional failover event, it still assumes the role of active processor. You can then troubleshoot the underlying cause and restart the processor as normal in the failover region.
Observability
Atlas Stream Processing provides various observability tools to assess the performance and health of your workspaces, connections, and processors. Additionally, to track regional service status and outages, you can visit the MongoDB Cloud Status page.
Scale Atlas Stream Processing
Atlas Stream Processing provisions resources per stream processor. To learn more, see the Tier Selection Guide.
For Atlas Stream Processing processors that operate on Atlas clusters, ensure you also scale your both your processors and clusters proportionate to your pipeline I/O needs. In particular, evaluate processors that perform a high volume of $lookup or $merge operations as candidates for cluster upscaling.
Change Stream Considerations
Atlas Stream Processing can use a Atlas cluster $source to consume change stream events. To ensure optimal performance and reliability for such stream processors, review the following considerations:
Oplog Window Size
Stream processors leverage the
$sourcecluster's oplog to resume service after stoppage. If a processor is down for a period greater than the oplog retention period, the resume token can point to truncated data, causing aChangeStreamHistoryLosterror.Mitigate this risk by aligning oplog sizing with acceptable downtime for your stream processing systems.
Resumption Options
Atlas Stream Processing supports resuming change-stream
$source``s from either the last stored checkpoint with ``resumeFromCheckpointor a specific time withstartAtOperationTime. To avoid gaps or duplications in processed data, define a standardized policy around which resumption options you use in a given scenario.Backlog Ingestion
Stream processors that operate on a large namespace can experience long initial periods of catch-up processing. Consider reducing the scope of your stream processor's namespace or choosing a more recent start point to reduce this initial catch-up period.
Change Stream Images
Auditing, compensating transactions, and undo operations require access to pre-processing document state. When you work with Atlas as a source, enable Document Pre- and Post-Images to support these operations. When you roll out support for document pre- or post-images, verify coverage and stability across your system.
The ``$replaceRoot`` Stage
Change stream sources return documents that bundle operational metadata with the content of the affected document. Writing these documents to a sink requires consumers to work with the change-event schema rather than the business logic of the source document. Unless you require the metadata, use the
$replaceRootstage to promote the source content subdocument before further processing.Join-Oriented Pipelines
Atlas Stream Processing supports the
$lookupstage to enable join-like enrichments of streaming data. These operations can increase processing latency. Consider limiting your Atlas Stream Processing pipelines to only those$lookupoperations that must be performed before writing to a sink, reserving more complex enrichments to precomputed views.Stale Checkpoints
Checkpoints contain only event history, not processing logic. When you resume operation from a checkpoint, your stream processor applies your current pipeline's logic to the old events, which can cause a mismatch between your existing downstream state and your new output. Consider introducing a new stream processor rather than resuming from a checkpoint if you require significantly different processing semantics.
Restart Planning
During restart, a stream processor might experience reduced throughput of new events. To increase system resiliency, schedule restarts to account for the temporarily reduced capacity.
Failover Planning
Develop standard procedures for sharing stream processor checkpoints across regions, choosing safe resumption points, and validating output consistency to ensure regional failover events don't lead to data quality issues.