[Fix] Permanently delete ERROR and TERMINATED state clusters if their creation fails (#4021)

## Changes
<!-- Summary of your changes that are easy to understand -->
If we get error or terminated cluster (after getting
WaitGetClusterRunning in Create) then we permanently delete them.

## Tests
<!-- 
How is this tested? Please see the checklist below and also describe any
other relevant tests
-->
Unit tests
- [ ] `make test` run locally
- [ ] relevant change in `docs/` folder
- [ ] covered with integration tests in `internal/acceptance`
- [ ] relevant acceptance tests are passing
- [ ] using Go SDK

Author: tanmay-db

clusters/resource_cluster.go

@@ -472,6 +472,11 @@ func resourceClusterCreate(ctx context.Context, d *schema.ResourceData, c *commo
 
 	clusterInfo, err := clusterWaiter.GetWithTimeout(timeout)
 	if err != nil {
+		// In case of "ERROR" or "TERMINATED" state, WaitGetClusterRunning returns an error and we should delete the cluster before returning
+		deleteError := resourceClusterDelete(ctx, d, c)
+		if deleteError != nil {
+			return fmt.Errorf("failed to create cluster: %v and failed to delete it during cleanup: %v", err, deleteError)
+		}
 		return err
 	}
 

clusters/resource_cluster_test.go

@@ -164,6 +164,117 @@ func TestResourceClusterCreatePinned(t *testing.T) {
 	assert.Equal(t, "abc", d.Id())
 }
 
+func TestResourceClusterCreateErrorFollowedByDeletion(t *testing.T) {
+	d, err := qa.ResourceFixture{
+		Fixtures: []qa.HTTPFixture{
+			{
+				Method:   "POST",
+				Resource: "/api/2.1/clusters/create",
+				ExpectedRequest: compute.CreateCluster{
+					NumWorkers:             100,
+					ClusterName:            "Shared Autoscaling",
+					SparkVersion:           "7.1-scala12",
+					NodeTypeId:             "i3.xlarge",
+					AutoterminationMinutes: 15,
+				},
+				Response: compute.ClusterDetails{
+					ClusterId: "abc",
+				},
+			},
+			{
+				Method:       "GET",
+				ReuseRequest: true,
+				Resource:     "/api/2.1/clusters/get?cluster_id=abc",
+				Response: compute.ClusterDetails{
+					ClusterId:              "abc",
+					NumWorkers:             100,
+					ClusterName:            "Shared Autoscaling",
+					SparkVersion:           "7.1-scala12",
+					NodeTypeId:             "i3.xlarge",
+					AutoterminationMinutes: 15,
+					State:                  compute.StateTerminated,
+				},
+			},
+			{
+				Method:   "POST",
+				Resource: "/api/2.1/clusters/permanent-delete",
+				ExpectedRequest: compute.PermanentDeleteCluster{
+					ClusterId: "abc",
+				},
+			},
+		},
+		Create:   true,
+		Resource: ResourceCluster(),
+		State: map[string]any{
+			"autotermination_minutes": 15,
+			"cluster_name":            "Shared Autoscaling",
+			"spark_version":           "7.1-scala12",
+			"node_type_id":            "i3.xlarge",
+			"num_workers":             100,
+		},
+	}.Apply(t)
+	assert.ErrorContains(t, err, "failed to reach RUNNING, got TERMINATED")
+	assert.Equal(t, "abc", d.Id())
+}
+
+func TestResourceClusterCreateErrorFollowedByDeletionError(t *testing.T) {
+	d, err := qa.ResourceFixture{
+		Fixtures: []qa.HTTPFixture{
+			{
+				Method:   "POST",
+				Resource: "/api/2.1/clusters/create",
+				ExpectedRequest: compute.CreateCluster{
+					NumWorkers:             100,
+					ClusterName:            "Shared Autoscaling",
+					SparkVersion:           "7.1-scala12",
+					NodeTypeId:             "i3.xlarge",
+					AutoterminationMinutes: 15,
+				},
+				Response: compute.ClusterDetails{
+					ClusterId: "abc",
+				},
+			},
+			{
+				Method:       "GET",
+				ReuseRequest: true,
+				Resource:     "/api/2.1/clusters/get?cluster_id=abc",
+				Response: compute.ClusterDetails{
+					ClusterId:              "abc",
+					NumWorkers:             100,
+					ClusterName:            "Shared Autoscaling",
+					SparkVersion:           "7.1-scala12",
+					NodeTypeId:             "i3.xlarge",
+					AutoterminationMinutes: 15,
+					State:                  compute.StateTerminated,
+				},
+			},
+			{
+				Method:   "POST",
+				Resource: "/api/2.1/clusters/permanent-delete",
+				ExpectedRequest: compute.PermanentDeleteCluster{
+					ClusterId: "abc",
+				},
+				Status: 500,
+				Response: common.APIErrorBody{
+					ErrorCode: "INTERNAL_ERROR",
+					Message:   "Internal error happened",
+				},
+			},
+		},
+		Create:   true,
+		Resource: ResourceCluster(),
+		State: map[string]any{
+			"autotermination_minutes": 15,
+			"cluster_name":            "Shared Autoscaling",
+			"spark_version":           "7.1-scala12",
+			"node_type_id":            "i3.xlarge",
+			"num_workers":             100,
+		},
+	}.Apply(t)
+	assert.ErrorContains(t, err, "failed to create cluster: failed to reach RUNNING, got TERMINATED:  and failed to delete it during cleanup: Internal error happened")
+	assert.Equal(t, "abc", d.Id())
+}
+
 func TestResourceClusterCreate_WithLibraries(t *testing.T) {
 	d, err := qa.ResourceFixture{
 		Fixtures: []qa.HTTPFixture{