feat: add Stream Processor resource

[sivaram-mongodb]

Proposed changes

Added new resource Stream Processor:

• Manages stream processing pipelines within Atlas Stream Processing workspaces
• Supports configurable stream aggregation pipelines with JSON-encoded stages
• Provides state management with automatic transitions (CREATED, STARTED, STOPPED)
• Includes Dead Letter Queue (DLQ) configuration for error handling
• Implements callback-based asynchronous operations for long-running processor creation
• Features configurable timeouts with optional cleanup on timeout
• Supports backward compatibility with both WorkspaceName and InstanceName fields
• Enables real-time data transformation and routing from sources to sinks

Resource Configuration:

The Stream Processor resource enables you to create and manage stream processing pipelines that transform and route data from source connections (Kafka, clusters, sample data) to sink destinations (clusters). Processors apply MongoDB aggregation pipeline stages to streaming data in real-time.

Required Properties:

• ProjectId: Atlas project identifier (24-hexadecimal characters)
• ProcessorName: Human-readable label for the stream processor
• Pipeline: JSON-encoded array of stream aggregation pipeline stages

Optional Properties:

• WorkspaceName or InstanceName: Stream workspace identifier (WorkspaceName is preferred)
• DesiredState: Target state for the processor (CREATED, STARTED, STOPPED)
• Options.Dlq: Dead Letter Queue configuration for failed messages
  • ConnectionName: Atlas cluster connection for DLQ storage
  • Db: Database name for DLQ messages
  • Coll: Collection name for DLQ messages
• Timeouts.Create: Timeout duration for creation operations (default: 20 minutes)
• DeleteOnCreateTimeout: Whether to delete the resource on timeout (default: true)
• Profile: AWS Secrets Manager profile (default: "default")

Read-Only Properties:

• Id: Unique 24-hexadecimal identifier
• State: Current processor state from Atlas API (CREATED, STARTED, STOPPED, FAILED)
• Stats: Processing statistics (available when processor is STARTED)

Create-Only Properties:

• ProjectId, ProcessorName, WorkspaceName, InstanceName, Profile: Cannot be changed after creation

cfn testing:

AWS Stacks:

Example 1 - Stream -> Cluster

Atlas CLI:

Before:

Create:

Update:

delete:

Example 2 - Stream -> Cluster (with DLQ)

Atlas CLI:

Before:

Create:

Update:

delete:

AWS Stacks:

Example 3 - Kafka -> Cluster

Atlas CLI:

Before:

Create:

Update:

delete:

AWS Stacks:

Example 4 - Stream -> Cluster

Atlas CLI:

Before:

Create:

Update:

delete:

Jira ticket: CLOUDP-368428

Please include a summary of the fix/feature/change, including any relevant motivation and context.

Link to any related issue(s):

Type of change:

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as
  expected)
• This change requires a documentation update
• If changes include removal or addition of 3rd party GitHub actions, I updated our internal document. Reach out to the APIx Integration slack channel to get access to the internal document.

Manual QA performed:

• cfn invoke for each of CRUDL/cfn test
• Updated resource in example
• Published to AWS private registry
• Used the template in example to create and update a stack in AWS
• Deleted stack to ensure resources are deleted
• Created multiple resources in same stack
• Validated in Atlas UI
• Included screenshots

Required Checklist:

• I have signed the MongoDB CLA
• I have added tests that prove my fix is effective or that my feature works
• I have checked that this change does not generate any credentials and that they are NOT accidentally logged anywhere.
• I have added any necessary documentation (if appropriate)
• I have run make fmt and formatted my code
• For CFN Resources: I have released by changes in the private registry and proved by change
  works in Atlas

Further comments

[maastha]

can we rename this file? not sure why this is called "share.go"? I see the main handlers here

[rakhul-mongo]

Yes, we've renamed it to a more appropriate name, 'handlers.go'. It was initially named 'share.go' following the pattern of flex-cluster. We assumed it was some sort of legacy name for handlers.

[maastha]

Can we update the description to callout the deprecation clearly?
Similar to TF messaging & referencing other CFN deprecated attributes:

"**WARNING:** This field is deprecated and will be removed in the next major release. Please use WorkspaceName instead" ...... "

[maastha]

Actually since this is a new resource, we should just remove "InstanceName" and we can include in Workspace Name description that this is same as 'InstanceName' in other stream resources
cc @marcosuma

[rakhul-mongo]

Valid point. We've removed InstanceName entirely since this is a new resource. We've also updated the WorkspaceName description.

[maastha]

common helper methods for such checks were added in a previous PR, can we merge that and then use those here before we merge this?

[rakhul-mongo]

Using common helper methods wherever applicable.

[maastha]

this would always return nil even on error

[rakhul-mongo]

Good catch. Fixed. The function now returns the error when cleanup fails, matching Terraform's behavior.

[oarbusi]

LGTM, thanks for addressing the comments