Sign in Dashboard Get started
  1. Registry
  2. Packages
  3. Mongodbatlas Provider
  4. API Docs
  5. StreamProcessor
Viewing docs for MongoDB Atlas v4.15.0
published on Friday, Aug 28, 2026 by Pulumi
Schema (JSON)
pulumi/pulumi-mongodbatlas

mongodbatlas.StreamProcessor

mongodbatlas logo mongodbatlas logo
Viewing docs for MongoDB Atlas v4.15.0
published on Friday, Aug 28, 2026 by Pulumi
Schema (JSON)
pulumi/pulumi-mongodbatlas

    mongodbatlas.StreamProcessor provides a Stream Processor resource. The resource lets you create, delete, import, start and stop a stream processor in a stream instance.

    NOTE: When updating an Atlas Stream Processor, the following behavior applies:

    1. If the processor is in a STARTED state, it will automatically be stopped before the update is applied
    2. The update will be performed while the processor is in STOPPED state
    3. If the processor was originally in STARTED state, it will be restarted after the update

    Pipeline field ordering

    IMPORTANT: MongoDB documents are ordered, and several pipeline constructs depend on the order of keys within a document: sort specifications, where key order is sort precedence; equality comparisons against a document literal, which match by exact field order; and $addFields/$project specifications, whose key order becomes the field order of the documents the processor writes. Do not build pipeline with jsonencode(): it emits object keys in lexicographic order, so a sort written as {"region": 1, "city": 1} reaches Atlas as {"city": 1, "region": 1}, reversing the sort precedence and silently changing what your processor does. Author pipeline as a raw JSON string instead — a heredoc, file("pipeline.json"), or templatefile("pipeline.json", { ... }) when the pipeline needs values interpolated into it — which Terraform passes through unchanged. Beware that jsonencode(jsondecode(...)), sometimes used to pull a pipeline out of a larger JSON document, sorts the keys for the same reason jsonencode() does; keep the pipeline in a file of its own so it can be read as a string. Note that pulumi preview still displays the attribute alphabetized and rendered as jsonencode(...); that is how Terraform renders any JSON-string attribute and does not reflect what is sent to Atlas. To confirm the order that was applied, inspect the request body with TF_LOG=DEBUG, or run sp.listStreamProcessors() against the workspace.

    Example Usage

    S

    import * as pulumi from "@pulumi/pulumi";
import * as mongodbatlas from "@pulumi/mongodbatlas";

const example = new mongodbatlas.StreamInstance("example", {
    projectId: projectId,
    instanceName: "InstanceName",
    dataProcessRegion: {
        region: "VIRGINIA_USA",
        cloudProvider: "AWS",
    },
});
const example_sample = new mongodbatlas.StreamConnection("example-sample", {
    projectId: projectId,
    workspaceName: example.instanceName,
    connectionName: "sample_stream_solar",
    type: "Sample",
});
const example_cluster = new mongodbatlas.StreamConnection("example-cluster", {
    projectId: projectId,
    workspaceName: example.instanceName,
    connectionName: "ClusterConnection",
    type: "Cluster",
    clusterName: clusterName,
    dbRoleToExecute: {
        role: "atlasAdmin",
        type: "BUILT_IN",
    },
});
const example_kafka = new mongodbatlas.StreamConnection("example-kafka", {
    projectId: projectId,
    workspaceName: example.instanceName,
    connectionName: "KafkaPlaintextConnection",
    type: "Kafka",
    authentication: {
        mechanism: "PLAIN",
        username: kafkaUsername,
        password: kafkaPassword,
    },
    bootstrapServers: "localhost:9092,localhost:9092",
    config: {
        "auto.offset.reset": "earliest",
    },
    security: {
        protocol: "SASL_PLAINTEXT",
    },
});
const stream_processor_sample_example = new mongodbatlas.StreamProcessor("stream-processor-sample-example", {
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: "sampleProcessorName",
    pipeline: pulumi.interpolate`[
  {\"$source\": {\"connectionName\": \"${example_sample.connectionName}\"}},
  {\"$emit\": {\"connectionName\": \"${example_cluster.connectionName}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
]
`,
    state: "STARTED",
    tier: "SP30",
});
const stream_processor_cluster_to_kafka_example = new mongodbatlas.StreamProcessor("stream-processor-cluster-to-kafka-example", {
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: "clusterProcessorName",
    pipeline: pulumi.interpolate`[
  {\"$source\": {\"connectionName\": \"${example_cluster.connectionName}\"}},
  {\"$emit\": {\"connectionName\": \"${example_kafka.connectionName}\", \"topic\": \"topic_from_cluster\"}}
]
`,
    state: "CREATED",
});
const stream_processor_kafka_to_cluster_example = new mongodbatlas.StreamProcessor("stream-processor-kafka-to-cluster-example", {
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: "kafkaProcessorName",
    pipeline: pulumi.interpolate`[
  {\"$source\": {\"connectionName\": \"${example_kafka.connectionName}\", \"topic\": \"topic_source\"}},
  {\"$emit\": {\"connectionName\": \"${example_cluster.connectionName}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
]
`,
    state: "CREATED",
    tier: "SP10",
    options: {
        dlq: {
            coll: "exampleColumn",
            connectionName: example_cluster.connectionName,
            db: "exampleDb",
        },
        autoscaling: {
            minTier: "SP10",
            maxTier: "SP50",
        },
    },
});
const example_stream_processors = mongodbatlas.getStreamProcessorsOutput({
    projectId: projectId,
    workspaceName: example.instanceName,
});
const example_stream_processor = mongodbatlas.getStreamProcessorOutput({
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: stream_processor_sample_example.processorName,
});
export const streamProcessorsState = example_stream_processor.state;
export const streamProcessorsResults = example_stream_processors.apply(example_stream_processors => example_stream_processors.results);

    import pulumi
import pulumi_mongodbatlas as mongodbatlas

example = mongodbatlas.StreamInstance("example",
    project_id=project_id,
    instance_name="InstanceName",
    data_process_region={
        "region": "VIRGINIA_USA",
        "cloud_provider": "AWS",
    })
example_sample = mongodbatlas.StreamConnection("example-sample",
    project_id=project_id,
    workspace_name=example.instance_name,
    connection_name="sample_stream_solar",
    type="Sample")
example_cluster = mongodbatlas.StreamConnection("example-cluster",
    project_id=project_id,
    workspace_name=example.instance_name,
    connection_name="ClusterConnection",
    type="Cluster",
    cluster_name=cluster_name,
    db_role_to_execute={
        "role": "atlasAdmin",
        "type": "BUILT_IN",
    })
example_kafka = mongodbatlas.StreamConnection("example-kafka",
    project_id=project_id,
    workspace_name=example.instance_name,
    connection_name="KafkaPlaintextConnection",
    type="Kafka",
    authentication={
        "mechanism": "PLAIN",
        "username": kafka_username,
        "password": kafka_password,
    },
    bootstrap_servers="localhost:9092,localhost:9092",
    config={
        "auto.offset.reset": "earliest",
    },
    security={
        "protocol": "SASL_PLAINTEXT",
    })
stream_processor_sample_example = mongodbatlas.StreamProcessor("stream-processor-sample-example",
    project_id=project_id,
    workspace_name=example.instance_name,
    processor_name="sampleProcessorName",
    pipeline=pulumi.Output.all(
        example-sampleConnection_name=example_sample.connection_name,
        example-clusterConnection_name=example_cluster.connection_name
).apply(lambda resolved_outputs: f"""[
  {{\"$source\": {{\"connectionName\": \"{resolved_outputs['example-sampleConnection_name']}\"}}}},
  {{\"$emit\": {{\"connectionName\": \"{resolved_outputs['example-clusterConnection_name']}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {{\"timeField\": \"_ts\"}}}}}}
]
""")
,
    state="STARTED",
    tier="SP30")
stream_processor_cluster_to_kafka_example = mongodbatlas.StreamProcessor("stream-processor-cluster-to-kafka-example",
    project_id=project_id,
    workspace_name=example.instance_name,
    processor_name="clusterProcessorName",
    pipeline=pulumi.Output.all(
        example-clusterConnection_name=example_cluster.connection_name,
        example-kafkaConnection_name=example_kafka.connection_name
).apply(lambda resolved_outputs: f"""[
  {{\"$source\": {{\"connectionName\": \"{resolved_outputs['example-clusterConnection_name']}\"}}}},
  {{\"$emit\": {{\"connectionName\": \"{resolved_outputs['example-kafkaConnection_name']}\", \"topic\": \"topic_from_cluster\"}}}}
]
""")
,
    state="CREATED")
stream_processor_kafka_to_cluster_example = mongodbatlas.StreamProcessor("stream-processor-kafka-to-cluster-example",
    project_id=project_id,
    workspace_name=example.instance_name,
    processor_name="kafkaProcessorName",
    pipeline=pulumi.Output.all(
        example-kafkaConnection_name=example_kafka.connection_name,
        example-clusterConnection_name=example_cluster.connection_name
).apply(lambda resolved_outputs: f"""[
  {{\"$source\": {{\"connectionName\": \"{resolved_outputs['example-kafkaConnection_name']}\", \"topic\": \"topic_source\"}}}},
  {{\"$emit\": {{\"connectionName\": \"{resolved_outputs['example-clusterConnection_name']}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {{\"timeField\": \"ts\"}}}}}}
]
""")
,
    state="CREATED",
    tier="SP10",
    options={
        "dlq": {
            "coll": "exampleColumn",
            "connection_name": example_cluster.connection_name,
            "db": "exampleDb",
        },
        "autoscaling": {
            "min_tier": "SP10",
            "max_tier": "SP50",
        },
    })
example_stream_processors = mongodbatlas.get_stream_processors_output(project_id=project_id,
    workspace_name=example.instance_name)
example_stream_processor = mongodbatlas.get_stream_processor_output(project_id=project_id,
    workspace_name=example.instance_name,
    processor_name=stream_processor_sample_example.processor_name)
pulumi.export("streamProcessorsState", example_stream_processor.state)
pulumi.export("streamProcessorsResults", example_stream_processors.results)

    package main

import (
	"fmt"

	"github.com/pulumi/pulumi-mongodbatlas/sdk/v4/go/mongodbatlas"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

func main() {
	pulumi.Run(func(ctx *pulumi.Context) error {
		example, err := mongodbatlas.NewStreamInstance(ctx, "example", &mongodbatlas.StreamInstanceArgs{
			ProjectId:    pulumi.Any(projectId),
			InstanceName: pulumi.String("InstanceName"),
			DataProcessRegion: &mongodbatlas.StreamInstanceDataProcessRegionArgs{
				Region:        pulumi.String("VIRGINIA_USA"),
				CloudProvider: pulumi.String("AWS"),
			},
		})
		if err != nil {
			return err
		}
		example_sample, err := mongodbatlas.NewStreamConnection(ctx, "example-sample", &mongodbatlas.StreamConnectionArgs{
			ProjectId:      pulumi.Any(projectId),
			WorkspaceName:  example.InstanceName,
			ConnectionName: pulumi.String("sample_stream_solar"),
			Type:           pulumi.String("Sample"),
		})
		if err != nil {
			return err
		}
		example_cluster, err := mongodbatlas.NewStreamConnection(ctx, "example-cluster", &mongodbatlas.StreamConnectionArgs{
			ProjectId:      pulumi.Any(projectId),
			WorkspaceName:  example.InstanceName,
			ConnectionName: pulumi.String("ClusterConnection"),
			Type:           pulumi.String("Cluster"),
			ClusterName:    pulumi.Any(clusterName),
			DbRoleToExecute: &mongodbatlas.StreamConnectionDbRoleToExecuteArgs{
				Role: pulumi.String("atlasAdmin"),
				Type: pulumi.String("BUILT_IN"),
			},
		})
		if err != nil {
			return err
		}
		example_kafka, err := mongodbatlas.NewStreamConnection(ctx, "example-kafka", &mongodbatlas.StreamConnectionArgs{
			ProjectId:      pulumi.Any(projectId),
			WorkspaceName:  example.InstanceName,
			ConnectionName: pulumi.String("KafkaPlaintextConnection"),
			Type:           pulumi.String("Kafka"),
			Authentication: &mongodbatlas.StreamConnectionAuthenticationArgs{
				Mechanism: pulumi.String("PLAIN"),
				Username:  pulumi.Any(kafkaUsername),
				Password:  pulumi.Any(kafkaPassword),
			},
			BootstrapServers: pulumi.String("localhost:9092,localhost:9092"),
			Config: pulumi.StringMap{
				"auto.offset.reset": pulumi.String("earliest"),
			},
			Security: &mongodbatlas.StreamConnectionSecurityArgs{
				Protocol: pulumi.String("SASL_PLAINTEXT"),
			},
		})
		if err != nil {
			return err
		}
		stream_processor_sample_example, err := mongodbatlas.NewStreamProcessor(ctx, "stream-processor-sample-example", &mongodbatlas.StreamProcessorArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: pulumi.String("sampleProcessorName"),
			Pipeline: pulumi.All(example_sample.ConnectionName, example_cluster.ConnectionName).ApplyT(func(_args []interface{}) (string, error) {
				example - sampleConnectionName := _args[0].(string)
				example - clusterConnectionName := _args[1].(string)
				return fmt.Sprintf("[\n  {\\\"$source\\\": {\\\"connectionName\\\": \\\"%v\\\"}},\n  {\\\"$emit\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"db\\\": \\\"sample\\\", \\\"coll\\\": \\\"solar\\\", \\\"timeseries\\\": {\\\"timeField\\\": \\\"_ts\\\"}}}\n]\n", example_sampleConnectionName, example_clusterConnectionName), nil
			}).(pulumi.StringOutput),
			State: pulumi.String("STARTED"),
			Tier:  pulumi.String("SP30"),
		})
		if err != nil {
			return err
		}
		_, err = mongodbatlas.NewStreamProcessor(ctx, "stream-processor-cluster-to-kafka-example", &mongodbatlas.StreamProcessorArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: pulumi.String("clusterProcessorName"),
			Pipeline: pulumi.All(example_cluster.ConnectionName, example_kafka.ConnectionName).ApplyT(func(_args []interface{}) (string, error) {
				example - clusterConnectionName := _args[0].(string)
				example - kafkaConnectionName := _args[1].(string)
				return fmt.Sprintf("[\n  {\\\"$source\\\": {\\\"connectionName\\\": \\\"%v\\\"}},\n  {\\\"$emit\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"topic\\\": \\\"topic_from_cluster\\\"}}\n]\n", example_clusterConnectionName, example_kafkaConnectionName), nil
			}).(pulumi.StringOutput),
			State: pulumi.String("CREATED"),
		})
		if err != nil {
			return err
		}
		_, err = mongodbatlas.NewStreamProcessor(ctx, "stream-processor-kafka-to-cluster-example", &mongodbatlas.StreamProcessorArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: pulumi.String("kafkaProcessorName"),
			Pipeline: pulumi.All(example_kafka.ConnectionName, example_cluster.ConnectionName).ApplyT(func(_args []interface{}) (string, error) {
				example - kafkaConnectionName := _args[0].(string)
				example - clusterConnectionName := _args[1].(string)
				return fmt.Sprintf("[\n  {\\\"$source\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"topic\\\": \\\"topic_source\\\"}},\n  {\\\"$emit\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"db\\\": \\\"kafka\\\", \\\"coll\\\": \\\"topic_source\\\", \\\"timeseries\\\": {\\\"timeField\\\": \\\"ts\\\"}}}\n]\n", example_kafkaConnectionName, example_clusterConnectionName), nil
			}).(pulumi.StringOutput),
			State: pulumi.String("CREATED"),
			Tier:  pulumi.String("SP10"),
			Options: &mongodbatlas.StreamProcessorOptionsArgs{
				Dlq: &mongodbatlas.StreamProcessorOptionsDlqArgs{
					Coll:           pulumi.String("exampleColumn"),
					ConnectionName: example_cluster.ConnectionName,
					Db:             pulumi.String("exampleDb"),
				},
				Autoscaling: &mongodbatlas.StreamProcessorOptionsAutoscalingArgs{
					MinTier: pulumi.String("SP10"),
					MaxTier: pulumi.String("SP50"),
				},
			},
		})
		if err != nil {
			return err
		}
		example_stream_processors := mongodbatlas.GetStreamProcessorsOutput(ctx, mongodbatlas.GetStreamProcessorsOutputArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
		}, nil)
		example_stream_processor := mongodbatlas.GetStreamProcessorOutput(ctx, mongodbatlas.GetStreamProcessorOutputArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: stream_processor_sample_example.ProcessorName,
		}, nil)
		ctx.Export("streamProcessorsState", example_stream_processor.State())
		ctx.Export("streamProcessorsResults", example_stream_processors.ApplyT(func(example_stream_processors mongodbatlas.GetStreamProcessorsResult) ([]mongodbatlas.GetStreamProcessorsResult, error) {
			return example_stream_processors.Results.([]mongodbatlas.GetStreamProcessorsResult), nil
		}).(pulumi.ArrayOutput))
		return nil
	})
}

    using System.Collections.Generic;
using System.Linq;
using Pulumi;
using Mongodbatlas = Pulumi.Mongodbatlas;

return await Deployment.RunAsync(() => 
{
    var example = new Mongodbatlas.StreamInstance("example", new()
    {
        ProjectId = projectId,
        InstanceName = "InstanceName",
        DataProcessRegion = new Mongodbatlas.Inputs.StreamInstanceDataProcessRegionArgs
        {
            Region = "VIRGINIA_USA",
            CloudProvider = "AWS",
        },
    });

    var example_sample = new Mongodbatlas.StreamConnection("example-sample", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ConnectionName = "sample_stream_solar",
        Type = "Sample",
    });

    var example_cluster = new Mongodbatlas.StreamConnection("example-cluster", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ConnectionName = "ClusterConnection",
        Type = "Cluster",
        ClusterName = clusterName,
        DbRoleToExecute = new Mongodbatlas.Inputs.StreamConnectionDbRoleToExecuteArgs
        {
            Role = "atlasAdmin",
            Type = "BUILT_IN",
        },
    });

    var example_kafka = new Mongodbatlas.StreamConnection("example-kafka", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ConnectionName = "KafkaPlaintextConnection",
        Type = "Kafka",
        Authentication = new Mongodbatlas.Inputs.StreamConnectionAuthenticationArgs
        {
            Mechanism = "PLAIN",
            Username = kafkaUsername,
            Password = kafkaPassword,
        },
        BootstrapServers = "localhost:9092,localhost:9092",
        Config = 
        {
            { "auto.offset.reset", "earliest" },
        },
        Security = new Mongodbatlas.Inputs.StreamConnectionSecurityArgs
        {
            Protocol = "SASL_PLAINTEXT",
        },
    });

    var stream_processor_sample_example = new Mongodbatlas.StreamProcessor("stream-processor-sample-example", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = "sampleProcessorName",
        Pipeline = Output.Tuple(example_sample.ConnectionName, example_cluster.ConnectionName).Apply(values =>
        {
            var example-sampleConnectionName = values.Item1;
            var example-clusterConnectionName = values.Item2;
            return @$"[
  {{\""$source\"": {{\""connectionName\"": \""{example_sampleConnectionName}\""}}}},
  {{\""$emit\"": {{\""connectionName\"": \""{example_clusterConnectionName}\"", \""db\"": \""sample\"", \""coll\"": \""solar\"", \""timeseries\"": {{\""timeField\"": \""_ts\""}}}}}}
]
";
        }),
        State = "STARTED",
        Tier = "SP30",
    });

    var stream_processor_cluster_to_kafka_example = new Mongodbatlas.StreamProcessor("stream-processor-cluster-to-kafka-example", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = "clusterProcessorName",
        Pipeline = Output.Tuple(example_cluster.ConnectionName, example_kafka.ConnectionName).Apply(values =>
        {
            var example-clusterConnectionName = values.Item1;
            var example-kafkaConnectionName = values.Item2;
            return @$"[
  {{\""$source\"": {{\""connectionName\"": \""{example_clusterConnectionName}\""}}}},
  {{\""$emit\"": {{\""connectionName\"": \""{example_kafkaConnectionName}\"", \""topic\"": \""topic_from_cluster\""}}}}
]
";
        }),
        State = "CREATED",
    });

    var stream_processor_kafka_to_cluster_example = new Mongodbatlas.StreamProcessor("stream-processor-kafka-to-cluster-example", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = "kafkaProcessorName",
        Pipeline = Output.Tuple(example_kafka.ConnectionName, example_cluster.ConnectionName).Apply(values =>
        {
            var example-kafkaConnectionName = values.Item1;
            var example-clusterConnectionName = values.Item2;
            return @$"[
  {{\""$source\"": {{\""connectionName\"": \""{example_kafkaConnectionName}\"", \""topic\"": \""topic_source\""}}}},
  {{\""$emit\"": {{\""connectionName\"": \""{example_clusterConnectionName}\"", \""db\"": \""kafka\"", \""coll\"": \""topic_source\"", \""timeseries\"": {{\""timeField\"": \""ts\""}}}}}}
]
";
        }),
        State = "CREATED",
        Tier = "SP10",
        Options = new Mongodbatlas.Inputs.StreamProcessorOptionsArgs
        {
            Dlq = new Mongodbatlas.Inputs.StreamProcessorOptionsDlqArgs
            {
                Coll = "exampleColumn",
                ConnectionName = example_cluster.ConnectionName,
                Db = "exampleDb",
            },
            Autoscaling = new Mongodbatlas.Inputs.StreamProcessorOptionsAutoscalingArgs
            {
                MinTier = "SP10",
                MaxTier = "SP50",
            },
        },
    });

    var example_stream_processors = Mongodbatlas.GetStreamProcessors.Invoke(new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
    });

    var example_stream_processor = Mongodbatlas.GetStreamProcessor.Invoke(new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = stream_processor_sample_example.ProcessorName,
    });

    return new Dictionary<string, object?>
    {
        ["streamProcessorsState"] = example_stream_processor.Apply(example_stream_processor => example_stream_processor.Apply(getStreamProcessorResult => getStreamProcessorResult.State)),
        ["streamProcessorsResults"] = example_stream_processors.Apply(example_stream_processors => example_stream_processors.Apply(getStreamProcessorsResult => getStreamProcessorsResult.Results)),
    };
});

    package generated_program;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.core.Output;
import com.pulumi.mongodbatlas.StreamInstance;
import com.pulumi.mongodbatlas.StreamInstanceArgs;
import com.pulumi.mongodbatlas.inputs.StreamInstanceDataProcessRegionArgs;
import com.pulumi.mongodbatlas.StreamConnection;
import com.pulumi.mongodbatlas.StreamConnectionArgs;
import com.pulumi.mongodbatlas.inputs.StreamConnectionDbRoleToExecuteArgs;
import com.pulumi.mongodbatlas.inputs.StreamConnectionAuthenticationArgs;
import com.pulumi.mongodbatlas.inputs.StreamConnectionSecurityArgs;
import com.pulumi.mongodbatlas.StreamProcessor;
import com.pulumi.mongodbatlas.StreamProcessorArgs;
import com.pulumi.mongodbatlas.inputs.StreamProcessorOptionsArgs;
import com.pulumi.mongodbatlas.inputs.StreamProcessorOptionsDlqArgs;
import com.pulumi.mongodbatlas.inputs.StreamProcessorOptionsAutoscalingArgs;
import com.pulumi.mongodbatlas.MongodbatlasFunctions;
import com.pulumi.mongodbatlas.inputs.GetStreamProcessorsArgs;
import com.pulumi.mongodbatlas.inputs.GetStreamProcessorArgs;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Map;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Paths;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        var example = new StreamInstance("example", StreamInstanceArgs.builder()
            .projectId(projectId)
            .instanceName("InstanceName")
            .dataProcessRegion(StreamInstanceDataProcessRegionArgs.builder()
                .region("VIRGINIA_USA")
                .cloudProvider("AWS")
                .build())
            .build());

        var example_sample = new StreamConnection("example-sample", StreamConnectionArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .connectionName("sample_stream_solar")
            .type("Sample")
            .build());

        var example_cluster = new StreamConnection("example-cluster", StreamConnectionArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .connectionName("ClusterConnection")
            .type("Cluster")
            .clusterName(clusterName)
            .dbRoleToExecute(StreamConnectionDbRoleToExecuteArgs.builder()
                .role("atlasAdmin")
                .type("BUILT_IN")
                .build())
            .build());

        var example_kafka = new StreamConnection("example-kafka", StreamConnectionArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .connectionName("KafkaPlaintextConnection")
            .type("Kafka")
            .authentication(StreamConnectionAuthenticationArgs.builder()
                .mechanism("PLAIN")
                .username(kafkaUsername)
                .password(kafkaPassword)
                .build())
            .bootstrapServers("localhost:9092,localhost:9092")
            .config(Map.of("auto.offset.reset", "earliest"))
            .security(StreamConnectionSecurityArgs.builder()
                .protocol("SASL_PLAINTEXT")
                .build())
            .build());

        var stream_processor_sample_example = new StreamProcessor("stream-processor-sample-example", StreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName("sampleProcessorName")
            .pipeline(Output.tuple(example_sample.connectionName(), example_cluster.connectionName()).applyValue(values -> {
                var example-sampleConnectionName = values.t1;
                var example-clusterConnectionName = values.t2;
                return """
[
  {\"$source\": {\"connectionName\": \"%s\"}},
  {\"$emit\": {\"connectionName\": \"%s\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
]
", example_sampleConnectionName,example_clusterConnectionName);
            }))
            .state("STARTED")
            .tier("SP30")
            .build());

        var stream_processor_cluster_to_kafka_example = new StreamProcessor("stream-processor-cluster-to-kafka-example", StreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName("clusterProcessorName")
            .pipeline(Output.tuple(example_cluster.connectionName(), example_kafka.connectionName()).applyValue(values -> {
                var example-clusterConnectionName = values.t1;
                var example-kafkaConnectionName = values.t2;
                return """
[
  {\"$source\": {\"connectionName\": \"%s\"}},
  {\"$emit\": {\"connectionName\": \"%s\", \"topic\": \"topic_from_cluster\"}}
]
", example_clusterConnectionName,example_kafkaConnectionName);
            }))
            .state("CREATED")
            .build());

        var stream_processor_kafka_to_cluster_example = new StreamProcessor("stream-processor-kafka-to-cluster-example", StreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName("kafkaProcessorName")
            .pipeline(Output.tuple(example_kafka.connectionName(), example_cluster.connectionName()).applyValue(values -> {
                var example-kafkaConnectionName = values.t1;
                var example-clusterConnectionName = values.t2;
                return """
[
  {\"$source\": {\"connectionName\": \"%s\", \"topic\": \"topic_source\"}},
  {\"$emit\": {\"connectionName\": \"%s\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
]
", example_kafkaConnectionName,example_clusterConnectionName);
            }))
            .state("CREATED")
            .tier("SP10")
            .options(StreamProcessorOptionsArgs.builder()
                .dlq(StreamProcessorOptionsDlqArgs.builder()
                    .coll("exampleColumn")
                    .connectionName(example_cluster.connectionName())
                    .db("exampleDb")
                    .build())
                .autoscaling(StreamProcessorOptionsAutoscalingArgs.builder()
                    .minTier("SP10")
                    .maxTier("SP50")
                    .build())
                .build())
            .build());

        final var example-stream-processors = MongodbatlasFunctions.getStreamProcessors(GetStreamProcessorsArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .build());

        final var example-stream-processor = MongodbatlasFunctions.getStreamProcessor(GetStreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName(stream_processor_sample_example.processorName())
            .build());

        ctx.export("streamProcessorsState", example_stream_processor.applyValue(_example_stream_processor -> _example_stream_processor.state()));
        ctx.export("streamProcessorsResults", example_stream_processors.applyValue(_example_stream_processors -> _example_stream_processors.results()));
    }
}

    resources:
  example:
    type: mongodbatlas:StreamInstance
    properties:
      projectId: ${projectId}
      instanceName: InstanceName
      dataProcessRegion:
        region: VIRGINIA_USA
        cloudProvider: AWS
  example-sample:
    type: mongodbatlas:StreamConnection
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      connectionName: sample_stream_solar
      type: Sample
  example-cluster:
    type: mongodbatlas:StreamConnection
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      connectionName: ClusterConnection
      type: Cluster
      clusterName: ${clusterName}
      dbRoleToExecute:
        role: atlasAdmin
        type: BUILT_IN
  example-kafka:
    type: mongodbatlas:StreamConnection
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      connectionName: KafkaPlaintextConnection
      type: Kafka
      authentication:
        mechanism: PLAIN
        username: ${kafkaUsername}
        password: ${kafkaPassword}
      bootstrapServers: localhost:9092,localhost:9092
      config:
        auto.offset.reset: earliest
      security:
        protocol: SASL_PLAINTEXT
  stream-processor-sample-example:
    type: mongodbatlas:StreamProcessor
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      processorName: sampleProcessorName
      pipeline: |
        [
          {\"$source\": {\"connectionName\": \"${["example-sample"].connectionName}\"}},
          {\"$emit\": {\"connectionName\": \"${["example-cluster"].connectionName}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
        ]
      state: STARTED
      tier: SP30
  stream-processor-cluster-to-kafka-example:
    type: mongodbatlas:StreamProcessor
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      processorName: clusterProcessorName
      pipeline: |
        [
          {\"$source\": {\"connectionName\": \"${["example-cluster"].connectionName}\"}},
          {\"$emit\": {\"connectionName\": \"${["example-kafka"].connectionName}\", \"topic\": \"topic_from_cluster\"}}
        ]
      state: CREATED
  stream-processor-kafka-to-cluster-example:
    type: mongodbatlas:StreamProcessor
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      processorName: kafkaProcessorName
      pipeline: |
        [
          {\"$source\": {\"connectionName\": \"${["example-kafka"].connectionName}\", \"topic\": \"topic_source\"}},
          {\"$emit\": {\"connectionName\": \"${["example-cluster"].connectionName}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
        ]
      state: CREATED
      tier: SP10
      options:
        dlq:
          coll: exampleColumn
          connectionName: ${["example-cluster"].connectionName}
          db: exampleDb
        autoscaling:
          minTier: SP10
          maxTier: SP50
variables:
  example-stream-processors:
    fn::invoke:
      function: mongodbatlas:getStreamProcessors
      arguments:
        projectId: ${projectId}
        workspaceName: ${example.instanceName}
  example-stream-processor:
    fn::invoke:
      function: mongodbatlas:getStreamProcessor
      arguments:
        projectId: ${projectId}
        workspaceName: ${example.instanceName}
        processorName: ${["stream-processor-sample-example"].processorName}
outputs:
  # example making use of data sources
  streamProcessorsState: ${["example-stream-processor"].state}
  streamProcessorsResults: ${["example-stream-processors"].results}

    pulumi {
  required_providers {
    mongodbatlas = {
      source = "pulumi/mongodbatlas"
    }
  }
}

data "mongodbatlas_getstreamprocessors" "example-stream-processors" {
  project_id     = projectId
  workspace_name = mongodbatlas_streaminstance.example.instance_name
}
data "mongodbatlas_getstreamprocessor" "example-stream-processor" {
  project_id     = projectId
  workspace_name = mongodbatlas_streaminstance.example.instance_name
  processor_name = mongodbatlas_streamprocessor.stream-processor-sample-example.processor_name
}

resource "mongodbatlas_streaminstance" "example" {
  project_id    = projectId
  instance_name = "InstanceName"
  data_process_region = {
    region         = "VIRGINIA_USA"
    cloud_provider = "AWS"
  }
}
resource "mongodbatlas_streamconnection" "example-sample" {
  project_id      = projectId
  workspace_name  = mongodbatlas_streaminstance.example.instance_name
  connection_name = "sample_stream_solar"
  type            = "Sample"
}
resource "mongodbatlas_streamconnection" "example-cluster" {
  project_id      = projectId
  workspace_name  = mongodbatlas_streaminstance.example.instance_name
  connection_name = "ClusterConnection"
  type            = "Cluster"
  cluster_name    = clusterName
  db_role_to_execute = {
    role = "atlasAdmin"
    type = "BUILT_IN"
  }
}
resource "mongodbatlas_streamconnection" "example-kafka" {
  project_id      = projectId
  workspace_name  = mongodbatlas_streaminstance.example.instance_name
  connection_name = "KafkaPlaintextConnection"
  type            = "Kafka"
  authentication = {
    mechanism = "PLAIN"
    username  = kafkaUsername
    password  = kafkaPassword
  }
  bootstrap_servers = "localhost:9092,localhost:9092"
  config = {
    "auto.offset.reset" = "earliest"
  }
  security = {
    protocol = "SASL_PLAINTEXT"
  }
}
resource "mongodbatlas_streamprocessor" "stream-processor-sample-example" {
  project_id     = projectId
  workspace_name = mongodbatlas_streaminstance.example.instance_name
  processor_name = "sampleProcessorName"
  pipeline       ="[
  {\"$source\": {\"connectionName\": \"${mongodbatlas_streamconnection.example-sample.connection_name}\"}},
  {\"$emit\": {\"connectionName\": \"${mongodbatlas_streamconnection.example-cluster.connection_name}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
]
"
  state          = "STARTED"
  tier           = "SP30"
}
resource "mongodbatlas_streamprocessor" "stream-processor-cluster-to-kafka-example" {
  project_id     = projectId
  workspace_name = mongodbatlas_streaminstance.example.instance_name
  processor_name = "clusterProcessorName"
  pipeline       ="[
  {\"$source\": {\"connectionName\": \"${mongodbatlas_streamconnection.example-cluster.connection_name}\"}},
  {\"$emit\": {\"connectionName\": \"${mongodbatlas_streamconnection.example-kafka.connection_name}\", \"topic\": \"topic_from_cluster\"}}
]
"
  state          = "CREATED"
}
resource "mongodbatlas_streamprocessor" "stream-processor-kafka-to-cluster-example" {
  project_id     = projectId
  workspace_name = mongodbatlas_streaminstance.example.instance_name
  processor_name = "kafkaProcessorName"
  pipeline       ="[
  {\"$source\": {\"connectionName\": \"${mongodbatlas_streamconnection.example-kafka.connection_name}\", \"topic\": \"topic_source\"}},
  {\"$emit\": {\"connectionName\": \"${mongodbatlas_streamconnection.example-cluster.connection_name}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
]
"
  state          = "CREATED"
  tier           = "SP10"
  options = {
    dlq = {
      coll            = "exampleColumn"
      connection_name = mongodbatlas_streamconnection.example-cluster.connection_name
      db              = "exampleDb"
    }
    autoscaling = {
      min_tier = "SP10"
      max_tier = "SP50"
    }
  }
}
# example making use of data sources
output "streamProcessorsState" {
  value = data.mongodbatlas_getstreamprocessor.example-stream-processor.state
}
output "streamProcessorsResults" {
  value = data.mongodbatlas_getstreamprocessors.example-stream-processors.results
}

    Further Examples

    • Atlas Stream Processor

    Create StreamProcessor Resource

    Resources are created with functions called constructors. To learn more about declaring and configuring resources, see Resources.

    Constructor syntax

    new StreamProcessor(name: string, args: StreamProcessorArgs, opts?: CustomResourceOptions);
    @overload
def StreamProcessor(resource_name: str,
                    args: StreamProcessorArgs,
                    opts: Optional[ResourceOptions] = None)

@overload
def StreamProcessor(resource_name: str,
                    opts: Optional[ResourceOptions] = None,
                    pipeline: Optional[str] = None,
                    processor_name: Optional[str] = None,
                    project_id: Optional[str] = None,
                    delete_on_create_timeout: Optional[bool] = None,
                    failover_enabled: Optional[bool] = None,
                    instance_name: Optional[str] = None,
                    options: Optional[StreamProcessorOptionsArgs] = None,
                    state: Optional[str] = None,
                    tier: Optional[str] = None,
                    timeouts: Optional[StreamProcessorTimeoutsArgs] = None,
                    workspace_name: Optional[str] = None)
    func NewStreamProcessor(ctx *Context, name string, args StreamProcessorArgs, opts ...ResourceOption) (*StreamProcessor, error)
    public StreamProcessor(string name, StreamProcessorArgs args, CustomResourceOptions? opts = null)
    public StreamProcessor(String name, StreamProcessorArgs args)
public StreamProcessor(String name, StreamProcessorArgs args, CustomResourceOptions options)

    type: mongodbatlas:StreamProcessor
properties: # The arguments to resource properties.
options: # Bag of options to control resource's behavior.


    resource "mongodbatlas_stream_processor" "name" {
    # resource properties
}

    Parameters

    name string
    The unique name of the resource.
    args StreamProcessorArgs
    The arguments to resource properties.
    opts CustomResourceOptions
    Bag of options to control resource's behavior.
    resource_name str
    The unique name of the resource.
    args StreamProcessorArgs
    The arguments to resource properties.
    opts ResourceOptions
    Bag of options to control resource's behavior.
    ctx Context
    Context object for the current deployment.
    name string
    The unique name of the resource.
    args StreamProcessorArgs
    The arguments to resource properties.
    opts ResourceOption
    Bag of options to control resource's behavior.
    name string
    The unique name of the resource.
    args StreamProcessorArgs
    The arguments to resource properties.
    opts CustomResourceOptions
    Bag of options to control resource's behavior.
    name String
    The unique name of the resource.
    args StreamProcessorArgs
    The arguments to resource properties.
    options CustomResourceOptions
    Bag of options to control resource's behavior.

    Constructor example

    The following reference example uses placeholder values for all input properties.

    var streamProcessorResource = new Mongodbatlas.StreamProcessor("streamProcessorResource", new()
{
    Pipeline = "string",
    ProcessorName = "string",
    ProjectId = "string",
    DeleteOnCreateTimeout = false,
    FailoverEnabled = false,
    Options = new Mongodbatlas.Inputs.StreamProcessorOptionsArgs
    {
        Autoscaling = new Mongodbatlas.Inputs.StreamProcessorOptionsAutoscalingArgs
        {
            MaxTier = "string",
            MinTier = "string",
        },
        Dlq = new Mongodbatlas.Inputs.StreamProcessorOptionsDlqArgs
        {
            Coll = "string",
            ConnectionName = "string",
            Db = "string",
        },
    },
    State = "string",
    Tier = "string",
    Timeouts = new Mongodbatlas.Inputs.StreamProcessorTimeoutsArgs
    {
        Create = "string",
    },
    WorkspaceName = "string",
});

    example, err := mongodbatlas.NewStreamProcessor(ctx, "streamProcessorResource", &mongodbatlas.StreamProcessorArgs{
	Pipeline:              pulumi.String("string"),
	ProcessorName:         pulumi.String("string"),
	ProjectId:             pulumi.String("string"),
	DeleteOnCreateTimeout: pulumi.Bool(false),
	FailoverEnabled:       pulumi.Bool(false),
	Options: &mongodbatlas.StreamProcessorOptionsArgs{
		Autoscaling: &mongodbatlas.StreamProcessorOptionsAutoscalingArgs{
			MaxTier: pulumi.String("string"),
			MinTier: pulumi.String("string"),
		},
		Dlq: &mongodbatlas.StreamProcessorOptionsDlqArgs{
			Coll:           pulumi.String("string"),
			ConnectionName: pulumi.String("string"),
			Db:             pulumi.String("string"),
		},
	},
	State: pulumi.String("string"),
	Tier:  pulumi.String("string"),
	Timeouts: &mongodbatlas.StreamProcessorTimeoutsArgs{
		Create: pulumi.String("string"),
	},
	WorkspaceName: pulumi.String("string"),
})

    resource "mongodbatlas_stream_processor" "streamProcessorResource" {
  lifecycle {
    create_before_destroy = true
  }
  pipeline                 = "string"
  processor_name           = "string"
  project_id               = "string"
  delete_on_create_timeout = false
  failover_enabled         = false
  options = {
    autoscaling = {
      max_tier = "string"
      min_tier = "string"
    }
    dlq = {
      coll            = "string"
      connection_name = "string"
      db              = "string"
    }
  }
  state = "string"
  tier  = "string"
  timeouts = {
    create = "string"
  }
  workspace_name = "string"
}

    var streamProcessorResource = new StreamProcessor("streamProcessorResource", StreamProcessorArgs.builder()
    .pipeline("string")
    .processorName("string")
    .projectId("string")
    .deleteOnCreateTimeout(false)
    .failoverEnabled(false)
    .options(StreamProcessorOptionsArgs.builder()
        .autoscaling(StreamProcessorOptionsAutoscalingArgs.builder()
            .maxTier("string")
            .minTier("string")
            .build())
        .dlq(StreamProcessorOptionsDlqArgs.builder()
            .coll("string")
            .connectionName("string")
            .db("string")
            .build())
        .build())
    .state("string")
    .tier("string")
    .timeouts(StreamProcessorTimeoutsArgs.builder()
        .create("string")
        .build())
    .workspaceName("string")
    .build());

    stream_processor_resource = mongodbatlas.StreamProcessor("streamProcessorResource",
    pipeline="string",
    processor_name="string",
    project_id="string",
    delete_on_create_timeout=False,
    failover_enabled=False,
    options={
        "autoscaling": {
            "max_tier": "string",
            "min_tier": "string",
        },
        "dlq": {
            "coll": "string",
            "connection_name": "string",
            "db": "string",
        },
    },
    state="string",
    tier="string",
    timeouts={
        "create": "string",
    },
    workspace_name="string")

    const streamProcessorResource = new mongodbatlas.StreamProcessor("streamProcessorResource", {
    pipeline: "string",
    processorName: "string",
    projectId: "string",
    deleteOnCreateTimeout: false,
    failoverEnabled: false,
    options: {
        autoscaling: {
            maxTier: "string",
            minTier: "string",
        },
        dlq: {
            coll: "string",
            connectionName: "string",
            db: "string",
        },
    },
    state: "string",
    tier: "string",
    timeouts: {
        create: "string",
    },
    workspaceName: "string",
});

    type: mongodbatlas:StreamProcessor
properties:
    deleteOnCreateTimeout: false
    failoverEnabled: false
    options:
        autoscaling:
            maxTier: string
            minTier: string
        dlq:
            coll: string
            connectionName: string
            db: string
    pipeline: string
    processorName: string
    projectId: string
    state: string
    tier: string
    timeouts:
        create: string
    workspaceName: string

    StreamProcessor Resource Properties

    To learn more about resource properties and how to use them, see Inputs and Outputs in the Architecture and Concepts docs.

    Inputs

    In Python, inputs that are objects can be passed either as argument classes or as dictionary literals.

    The StreamProcessor resource accepts the following input properties:

    Pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    ProcessorName string
    Label that identifies the stream processor.
    ProjectId string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    DeleteOnCreateTimeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    FailoverEnabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    InstanceName string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    Options StreamProcessorOptions
    Optional configuration for the stream processor. Empty options objects are not supported.
    State string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    Tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    Timeouts StreamProcessorTimeouts
    WorkspaceName string
    Label that identifies the stream processing workspace.
    Pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    ProcessorName string
    Label that identifies the stream processor.
    ProjectId string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    DeleteOnCreateTimeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    FailoverEnabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    InstanceName string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    Options StreamProcessorOptionsArgs
    Optional configuration for the stream processor. Empty options objects are not supported.
    State string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    Tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    Timeouts StreamProcessorTimeoutsArgs
    WorkspaceName string
    Label that identifies the stream processing workspace.
    pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processor_name string
    Label that identifies the stream processor.
    project_id string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    delete_on_create_timeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    failover_enabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instance_name string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options object
    Optional configuration for the stream processor. Empty options objects are not supported.
    state string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts object
    workspace_name string
    Label that identifies the stream processing workspace.
    pipeline String
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processorName String
    Label that identifies the stream processor.
    projectId String
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    deleteOnCreateTimeout Boolean
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    failoverEnabled Boolean
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instanceName String
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options StreamProcessorOptions
    Optional configuration for the stream processor. Empty options objects are not supported.
    state String
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    tier String
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts StreamProcessorTimeouts
    workspaceName String
    Label that identifies the stream processing workspace.
    pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processorName string
    Label that identifies the stream processor.
    projectId string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    deleteOnCreateTimeout boolean
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    failoverEnabled boolean
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instanceName string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options StreamProcessorOptions
    Optional configuration for the stream processor. Empty options objects are not supported.
    state string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts StreamProcessorTimeouts
    workspaceName string
    Label that identifies the stream processing workspace.
    pipeline str
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processor_name str
    Label that identifies the stream processor.
    project_id str
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    delete_on_create_timeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    failover_enabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instance_name str
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options StreamProcessorOptionsArgs
    Optional configuration for the stream processor. Empty options objects are not supported.
    state str
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    tier str
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts StreamProcessorTimeoutsArgs
    workspace_name str
    Label that identifies the stream processing workspace.
    pipeline String
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processorName String
    Label that identifies the stream processor.
    projectId String
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    deleteOnCreateTimeout Boolean
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    failoverEnabled Boolean
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instanceName String
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options Property Map
    Optional configuration for the stream processor. Empty options objects are not supported.
    state String
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    tier String
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts Property Map
    workspaceName String
    Label that identifies the stream processing workspace.

    Outputs

    All input properties are implicitly available as output properties. Additionally, the StreamProcessor resource produces the following output properties:

    EffectiveTier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    Id string
    The provider-assigned unique ID for this managed resource.
    Stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    EffectiveTier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    Id string
    The provider-assigned unique ID for this managed resource.
    Stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    effective_tier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    id string
    The provider-assigned unique ID for this managed resource.
    stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    effectiveTier String
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    id String
    The provider-assigned unique ID for this managed resource.
    stats String
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    effectiveTier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    id string
    The provider-assigned unique ID for this managed resource.
    stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    effective_tier str
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    id str
    The provider-assigned unique ID for this managed resource.
    stats str
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    effectiveTier String
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    id String
    The provider-assigned unique ID for this managed resource.
    stats String
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.

    Look up Existing StreamProcessor Resource

    Get an existing StreamProcessor resource’s state with the given name, ID, and optional extra properties used to qualify the lookup.

    public static get(name: string, id: Input<ID>, state?: StreamProcessorState, opts?: CustomResourceOptions): StreamProcessor
    @staticmethod
def get(resource_name: str,
        id: str,
        opts: Optional[ResourceOptions] = None,
        delete_on_create_timeout: Optional[bool] = None,
        effective_tier: Optional[str] = None,
        failover_enabled: Optional[bool] = None,
        instance_name: Optional[str] = None,
        options: Optional[StreamProcessorOptionsArgs] = None,
        pipeline: Optional[str] = None,
        processor_name: Optional[str] = None,
        project_id: Optional[str] = None,
        state: Optional[str] = None,
        stats: Optional[str] = None,
        tier: Optional[str] = None,
        timeouts: Optional[StreamProcessorTimeoutsArgs] = None,
        workspace_name: Optional[str] = None) -> StreamProcessor
    func GetStreamProcessor(ctx *Context, name string, id IDInput, state *StreamProcessorState, opts ...ResourceOption) (*StreamProcessor, error)
    public static StreamProcessor Get(string name, Input<string> id, StreamProcessorState? state, CustomResourceOptions? opts = null)
    public static StreamProcessor get(String name, Output<String> id, StreamProcessorState state, CustomResourceOptions options)
    resources:  _:    type: mongodbatlas:StreamProcessor    get:      id: ${id}
    import {
  to = mongodbatlas_stream_processor.example
  id = "${id}"
}
    name
    The unique name of the resulting resource.
    id
    The unique provider ID of the resource to lookup.
    state
    Any extra arguments used during the lookup.
    opts
    A bag of options that control this resource's behavior.
    resource_name
    The unique name of the resulting resource.
    id
    The unique provider ID of the resource to lookup.
    name
    The unique name of the resulting resource.
    id
    The unique provider ID of the resource to lookup.
    state
    Any extra arguments used during the lookup.
    opts
    A bag of options that control this resource's behavior.
    name
    The unique name of the resulting resource.
    id
    The unique provider ID of the resource to lookup.
    state
    Any extra arguments used during the lookup.
    opts
    A bag of options that control this resource's behavior.
    name
    The unique name of the resulting resource.
    id
    The unique provider ID of the resource to lookup.
    state
    Any extra arguments used during the lookup.
    opts
    A bag of options that control this resource's behavior.
    The following state arguments are supported:
    DeleteOnCreateTimeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    EffectiveTier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    FailoverEnabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    InstanceName string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    Options StreamProcessorOptions
    Optional configuration for the stream processor. Empty options objects are not supported.
    Pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    ProcessorName string
    Label that identifies the stream processor.
    ProjectId string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    State string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    Stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    Tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    Timeouts StreamProcessorTimeouts
    WorkspaceName string
    Label that identifies the stream processing workspace.
    DeleteOnCreateTimeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    EffectiveTier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    FailoverEnabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    InstanceName string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    Options StreamProcessorOptionsArgs
    Optional configuration for the stream processor. Empty options objects are not supported.
    Pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    ProcessorName string
    Label that identifies the stream processor.
    ProjectId string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    State string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    Stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    Tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    Timeouts StreamProcessorTimeoutsArgs
    WorkspaceName string
    Label that identifies the stream processing workspace.
    delete_on_create_timeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    effective_tier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    failover_enabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instance_name string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options object
    Optional configuration for the stream processor. Empty options objects are not supported.
    pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processor_name string
    Label that identifies the stream processor.
    project_id string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    state string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts object
    workspace_name string
    Label that identifies the stream processing workspace.
    deleteOnCreateTimeout Boolean
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    effectiveTier String
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    failoverEnabled Boolean
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instanceName String
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options StreamProcessorOptions
    Optional configuration for the stream processor. Empty options objects are not supported.
    pipeline String
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processorName String
    Label that identifies the stream processor.
    projectId String
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    state String
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    stats String
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    tier String
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts StreamProcessorTimeouts
    workspaceName String
    Label that identifies the stream processing workspace.
    deleteOnCreateTimeout boolean
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    effectiveTier string
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    failoverEnabled boolean
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instanceName string
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options StreamProcessorOptions
    Optional configuration for the stream processor. Empty options objects are not supported.
    pipeline string
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processorName string
    Label that identifies the stream processor.
    projectId string
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    state string
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    stats string
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    tier string
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts StreamProcessorTimeouts
    workspaceName string
    Label that identifies the stream processing workspace.
    delete_on_create_timeout bool
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    effective_tier str
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    failover_enabled bool
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instance_name str
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options StreamProcessorOptionsArgs
    Optional configuration for the stream processor. Empty options objects are not supported.
    pipeline str
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processor_name str
    Label that identifies the stream processor.
    project_id str
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    state str
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    stats str
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    tier str
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts StreamProcessorTimeoutsArgs
    workspace_name str
    Label that identifies the stream processing workspace.
    deleteOnCreateTimeout Boolean
    Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to true and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to false, the timeout will not trigger resource deletion. If you suspect a transient error when the value is true, wait before retrying to allow resource deletion to finish. Default is true.
    effectiveTier String
    Tier the stream processor is currently running on. When autoscaling is disabled this equals tier; when autoscaling is enabled it reflects the tier chosen by the autoscaler within the configured bounds.
    failoverEnabled Boolean
    Indicates whether this stream processor is eligible for failover. When true, an operator can trigger a failover event to migrate the stream processor to a secondary region configured in the workspace's failoverRegions. Requires an Atlas-to-Atlas or Atlas-to-Kafka pipeline with failoverRegions configured on the workspace.
    instanceName String
    Label that identifies the stream processing workspace.

    Deprecated: This parameter is deprecated. Please transition to workspace_name.

    options Property Map
    Optional configuration for the stream processor. Empty options objects are not supported.
    pipeline String
    Stream aggregation pipeline you want to apply to your streaming data, as a JSON string. MongoDB Atlas Docs contain more information. For more details see the Aggregation Pipelines Documentation. Field order matters: author this as a raw JSON string (heredoc or file("pipeline.json")) and do not use jsonencode, which sorts object keys lexicographically, changing sort precedence, document-literal equality matches, and $addFields/$project output field order.
    processorName String
    Label that identifies the stream processor.
    projectId String
    Unique 24-hexadecimal digit string that identifies your project, also known as groupId in the official documentation.
    state String
    The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are CREATED, STARTED or STOPPED. When a Stream Processor is created without specifying the state, it will default to CREATED state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
    stats String
    The stats associated with the stream processor. Refer to the MongoDB Atlas Docs for more information.
    tier String
    Selected tier to start a stream processor on rather than defaulting to the workspace setting. Configures Memory / VCPU allowances. Valid options are SP2, SP5, SP10, SP30, and SP50. When options.autoscaling is enabled, this is used only as the initial/baseline tier; the running tier is reported by effectiveTier.
    timeouts Property Map
    workspaceName String
    Label that identifies the stream processing workspace.

    Supporting Types

    StreamProcessorOptions, StreamProcessorOptionsArgs

    Autoscaling StreamProcessorOptionsAutoscaling
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    Dlq StreamProcessorOptionsDlq
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.
    Autoscaling StreamProcessorOptionsAutoscaling
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    Dlq StreamProcessorOptionsDlq
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.
    autoscaling object
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    dlq object
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.
    autoscaling StreamProcessorOptionsAutoscaling
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    dlq StreamProcessorOptionsDlq
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.
    autoscaling StreamProcessorOptionsAutoscaling
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    dlq StreamProcessorOptionsDlq
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.
    autoscaling StreamProcessorOptionsAutoscaling
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    dlq StreamProcessorOptionsDlq
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.
    autoscaling Property Map
    Vertical autoscaling configuration for the stream processor. When present, the processor automatically scales its tier between minTier and maxTier based on load; tier is used only as the initial/baseline tier and the running tier is reported by effectiveTier. To disable autoscaling, remove this block.
    dlq Property Map
    Dead letter queue for the stream processor. Refer to the MongoDB Atlas Docs for more information.

    StreamProcessorOptionsAutoscaling, StreamProcessorOptionsAutoscalingArgs

    MaxTier string
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    MinTier string
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.
    MaxTier string
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    MinTier string
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.
    max_tier string
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    min_tier string
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.
    maxTier String
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    minTier String
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.
    maxTier string
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    minTier string
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.
    max_tier str
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    min_tier str
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.
    maxTier String
    Tier ceiling for autoscaling (scale-up limit). When not set, it defaults to the workspace maximum tier.
    minTier String
    Tier floor for autoscaling (scale-down limit). When not set, it defaults to the lower of the processor tier and the workspace default tier.

    StreamProcessorOptionsDlq, StreamProcessorOptionsDlqArgs

    Coll string
    Name of the collection to use for the DLQ.
    ConnectionName string
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    Db string
    Name of the database to use for the DLQ.
    Coll string
    Name of the collection to use for the DLQ.
    ConnectionName string
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    Db string
    Name of the database to use for the DLQ.
    coll string
    Name of the collection to use for the DLQ.
    connection_name string
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    db string
    Name of the database to use for the DLQ.
    coll String
    Name of the collection to use for the DLQ.
    connectionName String
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    db String
    Name of the database to use for the DLQ.
    coll string
    Name of the collection to use for the DLQ.
    connectionName string
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    db string
    Name of the database to use for the DLQ.
    coll str
    Name of the collection to use for the DLQ.
    connection_name str
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    db str
    Name of the database to use for the DLQ.
    coll String
    Name of the collection to use for the DLQ.
    connectionName String
    Name of the connection to write DLQ messages to. Must be an Atlas connection.
    db String
    Name of the database to use for the DLQ.

    StreamProcessorTimeouts, StreamProcessorTimeoutsArgs

    Create string
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.
    Create string
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.
    create string
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.
    create String
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.
    create string
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.
    create str
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.
    create String
    A string that can be parsed as a duration consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), and "h" (hours). Default: 3h.

    Import

    Stream Processor resource can be imported using the Project ID, Stream Instance name and Stream Processor name, in the format INSTANCE_NAME-PROJECT_ID-PROCESSOR_NAME, e.g.

    $ terraform import mongodbatlas_stream_processor.test yourInstanceName-6117ac2fe2a3d04ed27a987v-yourProcessorName

    For more information see: MongoDB Atlas API - Stream Processor Documentation.

    To learn more about importing existing cloud resources, see Importing resources.

    Package Details

    Repository
    MongoDB Atlas pulumi/pulumi-mongodbatlas
    License
    Apache-2.0
    Notes
    This Pulumi package is based on the mongodbatlas Terraform Provider.
    mongodbatlas logo mongodbatlas logo
    Viewing docs for MongoDB Atlas v4.15.0
    published on Friday, Aug 28, 2026 by Pulumi
    Schema (JSON)
    pulumi/pulumi-mongodbatlas

      Try Pulumi Cloud free.
      Your team will thank you.

      Start free trial