feat: Long-running operation improvements for mongodbatlas_stream_processor resource (#3571)

* implement timeout on create

* delete_on_create_timeout

* delete_on_create_timeout test

* docs

* testing

Author: oarbusi

.changelog/3571.txt

@@ -0,0 +1,7 @@
+```release-note:enhancement
+resource/mongodbatlas_stream_processor: Adds `delete_on_create_timeout` attribute to indicate whether to delete the resource if its creation times out
+```
+
+```release-note:enhancement
+resource/mongodbatlas_stream_processor: Adds `timeouts` attribute for create operation
+```

docs/guides/2.0.0-upgrade-guide.md

@@ -21,7 +21,9 @@ The Terraform MongoDB Atlas Provider version 2.0.0 has the following new feature
   - `mongodbatlas_online_archive`
   - `mongodbatlas_privatelink_endpoint`
   - `mongodbatlas_privatelink_endpoint_service`
+  - `mongodbatlas_push_based_log_export`
   - `mongodbatlas_search_deployment`
+  - `mongodbatlas_stream_processor`
 
 ## Breaking Changes
 

docs/resources/stream_processor.md

@@ -132,10 +132,12 @@ output "stream_processors_results" {
 
 ### Optional
 
+- `delete_on_create_timeout` (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`.
 - `options` (Attributes) Optional configuration for the stream processor. (see [below for nested schema](#nestedatt--options))
 - `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. 
 
 **NOTE** When a Stream Processor is updated without specifying the state, it is stopped and then restored to previous state upon update completion.
+- `timeouts` (Attributes) (see [below for nested schema](#nestedatt--timeouts))
 
 ### Read-Only
 
@@ -158,6 +160,15 @@ Required:
 - `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.
 
+
+
+<a id="nestedatt--timeouts"></a>
+### Nested Schema for `timeouts`
+
+Optional:
+
+- `create` (String) A string that can be [parsed as a duration](https://pkg.go.dev/time#ParseDuration) consisting of numbers and unit suffixes, such as "30s" or "2h45m". Valid time units are "s" (seconds), "m" (minutes), "h" (hours).
+
 ## 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.
 ```

internal/service/streamprocessor/model.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 
 	"github.com/hashicorp/terraform-plugin-framework-jsontypes/jsontypes"
+	"github.com/hashicorp/terraform-plugin-framework-timeouts/resource/timeouts"
 	"github.com/hashicorp/terraform-plugin-framework/diag"
 	"github.com/hashicorp/terraform-plugin-framework/types"
 	"github.com/hashicorp/terraform-plugin-framework/types/basetypes"
@@ -79,7 +80,7 @@ func NewStreamProcessorUpdateReq(ctx context.Context, plan *TFStreamProcessorRSM
 	return streamProcessorAPIParams, nil
 }
 
-func NewStreamProcessorWithStats(ctx context.Context, projectID, instanceName string, apiResp *admin.StreamsProcessorWithStats) (*TFStreamProcessorRSModel, diag.Diagnostics) {
+func NewStreamProcessorWithStats(ctx context.Context, projectID, instanceName string, apiResp *admin.StreamsProcessorWithStats, timeout *timeouts.Value, deleteOnCreateTimeout *types.Bool) (*TFStreamProcessorRSModel, diag.Diagnostics) {
 	if apiResp == nil {
 		return nil, diag.Diagnostics{diag.NewErrorDiagnostic("streamProcessor API response is nil", "")}
 	}
@@ -105,6 +106,12 @@ func NewStreamProcessorWithStats(ctx context.Context, projectID, instanceName st
 		State:         types.StringPointerValue(&apiResp.State),
 		Stats:         statsTF,
 	}
+	if timeout != nil {
+		tfModel.Timeouts = *timeout
+	}
+	if deleteOnCreateTimeout != nil {
+		tfModel.DeleteOnCreateTimeout = *deleteOnCreateTimeout
+	}
 	return tfModel, nil
 }
 

internal/service/streamprocessor/model_test.go

@@ -230,7 +230,7 @@ func TestSDKToTFModel(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			sdkModel := tc.sdkModel
-			resultModel, diags := streamprocessor.NewStreamProcessorWithStats(t.Context(), projectID, instanceName, sdkModel)
+			resultModel, diags := streamprocessor.NewStreamProcessorWithStats(t.Context(), projectID, instanceName, sdkModel, nil, nil)
 			if diags.HasError() {
 				t.Fatalf("unexpected errors found: %s", diags.Errors()[0].Summary())
 			}

internal/service/streamprocessor/resource.go

@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform-plugin-framework/path"
 	"github.com/hashicorp/terraform-plugin-framework/resource"
 
+	"github.com/mongodb/terraform-provider-mongodbatlas/internal/common/cleanup"
 	"github.com/mongodb/terraform-provider-mongodbatlas/internal/common/conversion"
 	"github.com/mongodb/terraform-provider-mongodbatlas/internal/common/validate"
 	"github.com/mongodb/terraform-provider-mongodbatlas/internal/config"
@@ -86,7 +87,19 @@ func (r *streamProcessorRS) Create(ctx context.Context, req resource.CreateReque
 		ProcessorName: processorName,
 	}
 
-	streamProcessorResp, err := WaitStateTransition(ctx, streamProcessorParams, connV2.StreamsApi, []string{InitiatingState, CreatingState}, []string{CreatedState})
+	createTimeout := cleanup.ResolveTimeout(ctx, &plan.Timeouts, cleanup.OperationCreate, &resp.Diagnostics)
+	if resp.Diagnostics.HasError() {
+		return
+	}
+
+	streamProcessorResp, err := WaitStateTransitionWithTimeout(ctx, streamProcessorParams, connV2.StreamsApi, []string{InitiatingState, CreatingState}, []string{CreatedState}, createTimeout)
+	err = cleanup.HandleCreateTimeout(cleanup.ResolveDeleteOnCreateTimeout(plan.DeleteOnCreateTimeout), err, func(ctxCleanup context.Context) error {
+		_, err := connV2.StreamsApi.DeleteStreamProcessor(ctxCleanup, projectID, instanceName, processorName).Execute()
+		if err != nil {
+			return err
+		}
+		return nil
+	})
 	if err != nil {
 		resp.Diagnostics.AddError("Error creating stream processor", err.Error())
 		return
@@ -111,7 +124,7 @@ func (r *streamProcessorRS) Create(ctx context.Context, req resource.CreateReque
 		}
 	}
 
-	newStreamProcessorModel, diags := NewStreamProcessorWithStats(ctx, projectID, instanceName, streamProcessorResp)
+	newStreamProcessorModel, diags := NewStreamProcessorWithStats(ctx, projectID, instanceName, streamProcessorResp, &plan.Timeouts, &plan.DeleteOnCreateTimeout)
 	if diags.HasError() {
 		resp.Diagnostics.Append(diags...)
 		return
@@ -140,7 +153,7 @@ func (r *streamProcessorRS) Read(ctx context.Context, req resource.ReadRequest,
 		return
 	}
 
-	newStreamProcessorModel, diags := NewStreamProcessorWithStats(ctx, projectID, instanceName, streamProcessor)
+	newStreamProcessorModel, diags := NewStreamProcessorWithStats(ctx, projectID, instanceName, streamProcessor, &state.Timeouts, &state.DeleteOnCreateTimeout)
 	if diags.HasError() {
 		resp.Diagnostics.Append(diags...)
 		return
@@ -238,7 +251,7 @@ func (r *streamProcessorRS) Update(ctx context.Context, req resource.UpdateReque
 		}
 	}
 
-	newStreamProcessorModel, diags := NewStreamProcessorWithStats(ctx, projectID, instanceName, streamProcessorResp)
+	newStreamProcessorModel, diags := NewStreamProcessorWithStats(ctx, projectID, instanceName, streamProcessorResp, &plan.Timeouts, &plan.DeleteOnCreateTimeout)
 	if diags.HasError() {
 		resp.Diagnostics.Append(diags...)
 		return

internal/service/streamprocessor/resource_schema.go

@@ -4,11 +4,13 @@ import (
 	"context"
 
 	"github.com/hashicorp/terraform-plugin-framework-jsontypes/jsontypes"
+	"github.com/hashicorp/terraform-plugin-framework-timeouts/resource/timeouts"
 	"github.com/hashicorp/terraform-plugin-framework/attr"
 	"github.com/hashicorp/terraform-plugin-framework/resource/schema"
 	"github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier"
 	"github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier"
 	"github.com/hashicorp/terraform-plugin-framework/types"
+	"github.com/mongodb/terraform-provider-mongodbatlas/internal/common/customplanmodifier"
 )
 
 func ResourceSchema(ctx context.Context) schema.Schema {
@@ -73,19 +75,31 @@ func ResourceSchema(ctx context.Context) schema.Schema {
 				Computed:            true,
 				MarkdownDescription: "The stats associated with the stream processor. Refer to the [MongoDB Atlas Docs](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor/#view-statistics-of-a-stream-processor) for more information.",
 			},
+			"timeouts": timeouts.Attributes(ctx, timeouts.Opts{
+				Create: true,
+			}),
+			"delete_on_create_timeout": schema.BoolAttribute{
+				Optional: true,
+				PlanModifiers: []planmodifier.Bool{
+					customplanmodifier.CreateOnlyBoolPlanModifier(),
+				},
+				MarkdownDescription: "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`.",
+			},
 		},
 	}
 }
 
 type TFStreamProcessorRSModel struct {
-	InstanceName  types.String         `tfsdk:"instance_name"`
-	Options       types.Object         `tfsdk:"options"`
-	Pipeline      jsontypes.Normalized `tfsdk:"pipeline"`
-	ProcessorID   types.String         `tfsdk:"id"`
-	ProcessorName types.String         `tfsdk:"processor_name"`
-	ProjectID     types.String         `tfsdk:"project_id"`
-	State         types.String         `tfsdk:"state"`
-	Stats         types.String         `tfsdk:"stats"`
+	InstanceName          types.String         `tfsdk:"instance_name"`
+	Options               types.Object         `tfsdk:"options"`
+	Pipeline              jsontypes.Normalized `tfsdk:"pipeline"`
+	ProcessorID           types.String         `tfsdk:"id"`
+	ProcessorName         types.String         `tfsdk:"processor_name"`
+	ProjectID             types.String         `tfsdk:"project_id"`
+	State                 types.String         `tfsdk:"state"`
+	Stats                 types.String         `tfsdk:"stats"`
+	Timeouts              timeouts.Value       `tfsdk:"timeouts"`
+	DeleteOnCreateTimeout types.Bool           `tfsdk:"delete_on_create_timeout"`
 }
 
 type TFOptionsModel struct {

internal/service/streamprocessor/resource_test.go

@@ -57,12 +57,12 @@ func basicTestCase(t *testing.T) *resource.TestCase {
 		CheckDestroy:             checkDestroyStreamProcessor,
 		Steps: []resource.TestStep{
 			{
-				Config:            config(t, projectID, instanceName, processorName, "", randomSuffix, sampleSrcConfig, testLogDestConfig),
+				Config:            config(t, projectID, instanceName, processorName, "", randomSuffix, sampleSrcConfig, testLogDestConfig, "", nil),
 				Check:             composeStreamProcessorChecks(projectID, instanceName, processorName, streamprocessor.CreatedState, false, false),
 				ConfigStateChecks: pluralConfigStateChecks(processorName, streamprocessor.CreatedState, instanceName, false, false),
 			},
 			{
-				Config:            config(t, projectID, instanceName, processorName, streamprocessor.StartedState, randomSuffix, sampleSrcConfig, testLogDestConfig),
+				Config:            config(t, projectID, instanceName, processorName, streamprocessor.StartedState, randomSuffix, sampleSrcConfig, testLogDestConfig, "", nil),
 				Check:             composeStreamProcessorChecks(projectID, instanceName, processorName, streamprocessor.StartedState, true, false),
 				ConfigStateChecks: pluralConfigStateChecks(processorName, streamprocessor.StartedState, instanceName, true, false),
 			},
@@ -89,7 +89,7 @@ func TestAccStreamProcessor_JSONWhiteSpaceFormat(t *testing.T) {
 		CheckDestroy:             checkDestroyStreamProcessor,
 		Steps: []resource.TestStep{
 			{
-				Config:            config(t, projectID, instanceName, processorName, streamprocessor.CreatedState, randomSuffix, sampleSrcConfigExtraSpaces, testLogDestConfig),
+				Config:            config(t, projectID, instanceName, processorName, streamprocessor.CreatedState, randomSuffix, sampleSrcConfigExtraSpaces, testLogDestConfig, "", nil),
 				Check:             composeStreamProcessorChecks(projectID, instanceName, processorName, streamprocessor.CreatedState, false, false),
 				ConfigStateChecks: pluralConfigStateChecks(processorName, streamprocessor.CreatedState, instanceName, false, false),
 			},
@@ -112,7 +112,7 @@ func TestAccStreamProcessor_withOptions(t *testing.T) {
 		CheckDestroy:             checkDestroyStreamProcessor,
 		Steps: []resource.TestStep{
 			{
-				Config:            config(t, projectID, instanceName, processorName, streamprocessor.CreatedState, randomSuffix, src, dest),
+				Config:            config(t, projectID, instanceName, processorName, streamprocessor.CreatedState, randomSuffix, src, dest, "", nil),
 				Check:             composeStreamProcessorChecks(projectID, instanceName, processorName, streamprocessor.CreatedState, false, true),
 				ConfigStateChecks: pluralConfigStateChecks(processorName, streamprocessor.CreatedState, instanceName, false, true),
 			},
@@ -276,7 +276,7 @@ func TestAccStreamProcessor_clusterType(t *testing.T) {
 		CheckDestroy:             checkDestroyStreamProcessor,
 		Steps: []resource.TestStep{
 			{
-				Config:            config(t, projectID, instanceName, processorName, streamprocessor.StartedState, randomSuffix, srcConfig, testLogDestConfig),
+				Config:            config(t, projectID, instanceName, processorName, streamprocessor.StartedState, randomSuffix, srcConfig, testLogDestConfig, "", nil),
 				Check:             composeStreamProcessorChecks(projectID, instanceName, processorName, streamprocessor.StartedState, true, false),
 				ConfigStateChecks: pluralConfigStateChecks(processorName, streamprocessor.StartedState, instanceName, true, false),
 			},
@@ -297,16 +297,38 @@ func TestAccStreamProcessor_createErrors(t *testing.T) {
 		CheckDestroy:             checkDestroyStreamProcessor,
 		Steps: []resource.TestStep{
 			{
-				Config:      config(t, projectID, instanceName, processorName, streamprocessor.StoppedState, randomSuffix, invalidJSONConfig, testLogDestConfig),
+				Config:      config(t, projectID, instanceName, processorName, streamprocessor.StoppedState, randomSuffix, invalidJSONConfig, testLogDestConfig, "", nil),
 				ExpectError: regexp.MustCompile("Invalid JSON String Value"),
 			},
 			{
-				Config:      config(t, projectID, instanceName, processorName, streamprocessor.StoppedState, randomSuffix, sampleSrcConfig, testLogDestConfig),
+				Config:      config(t, projectID, instanceName, processorName, streamprocessor.StoppedState, randomSuffix, sampleSrcConfig, testLogDestConfig, "", nil),
 				ExpectError: regexp.MustCompile("When creating a stream processor, the only valid states are CREATED and STARTED"),
 			},
 		}})
 }
 
+func TestAccStreamProcessor_createTimeoutWithDeleteOnCreate(t *testing.T) {
+	acc.SkipTestForCI(t) // Creation of stream processor for testing is too fast to force the creation timeout
+	var (
+		projectID, instanceName = acc.ProjectIDExecutionWithStreamInstance(t)
+		processorName           = "new-processor"
+		randomSuffix            = acctest.RandString(5)
+		createTimeout           = "1s"
+		deleteOnCreateTimeout   = true
+	)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acc.PreCheckBasic(t) },
+		ProtoV6ProviderFactories: acc.TestAccProviderV6Factories,
+		CheckDestroy:             checkDestroyStreamProcessor,
+		Steps: []resource.TestStep{
+			{
+				Config:      config(t, projectID, instanceName, processorName, streamprocessor.StartedState, randomSuffix, sampleSrcConfig, testLogDestConfig, acc.TimeoutConfig(&createTimeout, nil, nil, true), &deleteOnCreateTimeout),
+				ExpectError: regexp.MustCompile("will run cleanup because delete_on_create_timeout is true"),
+			},
+		}})
+}
+
 func checkExists(resourceName string) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		rs, ok := s.RootModule().Resources[resourceName]
@@ -526,12 +548,16 @@ func composeStreamProcessorChecks(projectID, instanceName, processorName, state
 	return resource.ComposeAggregateTestCheckFunc(checks...)
 }
 
-func config(t *testing.T, projectID, instanceName, processorName, state, nameSuffix string, src, dest connectionConfig) string {
+func config(t *testing.T, projectID, instanceName, processorName, state, nameSuffix string, src, dest connectionConfig, timeoutConfig string, deleteOnCreateTimeout *bool) string {
 	t.Helper()
 	stateConfig := ""
 	if state != "" {
 		stateConfig = fmt.Sprintf(`state = %[1]q`, state)
 	}
+	deleteOnCreateTimeoutConfig := ""
+	if deleteOnCreateTimeout != nil {
+		deleteOnCreateTimeoutConfig = fmt.Sprintf(`delete_on_create_timeout = %[1]t`, *deleteOnCreateTimeout)
+	}
 
 	connectionConfigSrc, connectionIDSrc, pipelineStepSrc := configConnection(t, projectID, instanceName, src, nameSuffix)
 	connectionConfigDest, connectionIDDest, pipelineStepDest := configConnection(t, projectID, instanceName, dest, nameSuffix)
@@ -581,9 +607,11 @@ func config(t *testing.T, projectID, instanceName, processorName, state, nameSuf
 		%[5]s
 		%[6]s
 		depends_on = [%[7]s]
+		%[8]s
+		%[9]s
 		}
 		
-	`, projectID, instanceName, processorName, pipeline, stateConfig, optionsStr, dependsOnStr) + otherConfig
+	`, projectID, instanceName, processorName, pipeline, stateConfig, optionsStr, dependsOnStr, timeoutConfig, deleteOnCreateTimeoutConfig) + otherConfig
 }
 
 func configConnection(t *testing.T, projectID, instanceName string, config connectionConfig, nameSuffix string) (connectionConfig, resourceID, pipelineStep string) {

internal/service/streamprocessor/state_transition.go

@@ -24,15 +24,21 @@ const (
 const (
 	ErrorUpdateStateTransition = "Stream Processor must be in %s state to transition to %s state"
 	ErrorUpdateToCreatedState  = "Stream Processor cannot transition from %s to CREATED"
+	defaultTimeout             = 5 * time.Minute // big pipelines can take a while to stop due to checkpointing. By default, we prefer the API to raise the error (~ 3min) than having to expose custom timeouts.
+	minTimeout                 = 3 * time.Second
 )
 
 func WaitStateTransition(ctx context.Context, requestParams *admin.GetStreamProcessorApiParams, client admin.StreamsApi, pendingStates, desiredStates []string) (*admin.StreamsProcessorWithStats, error) {
+	return WaitStateTransitionWithTimeout(ctx, requestParams, client, pendingStates, desiredStates, defaultTimeout)
+}
+
+func WaitStateTransitionWithTimeout(ctx context.Context, requestParams *admin.GetStreamProcessorApiParams, client admin.StreamsApi, pendingStates, desiredStates []string, timeout time.Duration) (*admin.StreamsProcessorWithStats, error) {
 	stateConf := &retry.StateChangeConf{
 		Pending:    pendingStates,
 		Target:     desiredStates,
 		Refresh:    refreshFunc(ctx, requestParams, client),
-		Timeout:    5 * time.Minute, // big pipelines can take a while to stop due to checkpointing. We prefer the API to raise the error (~ 3min) than having to expose custom timeouts.
-		MinTimeout: 3 * time.Second,
+		Timeout:    timeout,
+		MinTimeout: minTimeout,
 		Delay:      0,
 	}