streaming_sink: deprecate cloud_provider and region fields

docs/resources/streaming_sink.md

@@ -77,11 +77,9 @@ resource "astra_streaming_sink" "streaming_sink" {
 ### Required
 
 - `auto_ack` (Boolean) auto ack
-- `cloud_provider` (String) Cloud provider
 - `namespace` (String) Pulsar Namespace
 - `parallelism` (Number) Parallelism for Pulsar sink
 - `processing_guarantees` (String) "ATLEAST_ONCE""ATMOST_ONCE""EFFECTIVELY_ONCE".
-- `region` (String) cloud region
 - `retain_ordering` (Boolean) Retain ordering.
 - `sink_configs` (String) Sink Configs
 - `sink_name` (String) Name of the sink.
@@ -91,8 +89,10 @@ resource "astra_streaming_sink" "streaming_sink" {
 ### Optional
 
 - `archive` (String) Name of the sink archive type to use. Defaults to the value of sink_name.  Must be formatted as a URL, e.g. 'builtin://jdbc-clickhouse
+- `cloud_provider` (String, Deprecated) Cloud provider
 - `deletion_protection` (Boolean) Whether or not to allow Terraform to destroy this streaming sink. Unless this field is set to false in Terraform state, a `terraform destroy` or `terraform apply` command that deletes the instance will fail. Defaults to `true`.
 - `pulsar_cluster` (String) Name of the pulsar cluster in which to create the sink.  If left blank, the name will be inferred from thecloud provider and region
+- `region` (String, Deprecated) cloud region
 
 ### Read-Only
 

internal/provider/resource_streaming_sink.go

@@ -54,15 +54,17 @@ func resourceStreamingSink() *schema.Resource {
 			},
 			"region": {
 				Description:  "cloud region",
+				Deprecated:   "use `pulsar_cluster` instead",
 				Type:         schema.TypeString,
-				Required:     true,
+				Optional:     true,
 				ForceNew:     true,
 				ValidateFunc: validation.StringMatch(regexp.MustCompile("^.{2,}"), "name must be atleast 2 characters"),
 			},
 			"cloud_provider": {
 				Description:  "Cloud provider",
+				Deprecated:   "use `pulsar_cluster` instead",
 				Type:         schema.TypeString,
-				Required:     true,
+				Optional:     true,
 				ForceNew:     true,
 				ValidateFunc: validation.StringMatch(regexp.MustCompile("^.{2,}"), "name must be atleast 2 characters"),
 			},
@@ -144,12 +146,17 @@ func resourceStreamingSinkDelete(ctx context.Context, resourceData *schema.Resou
 	sinkName := resourceData.Get("sink_name").(string)
 	namespace := resourceData.Get("namespace").(string)
 
-	pulsarClusterName := resourceData.Get("pulsar_cluster").(string)
+	pulsarClusterFromConfig := resourceData.Get("pulsar_cluster").(string)
 	rawRegion := resourceData.Get("region").(string)
 	region := strings.ReplaceAll(rawRegion, "-", "")
 	cloudProvider := resourceData.Get("cloud_provider").(string)
 
-	pulsarCluster := getPulsarCluster(pulsarClusterName, cloudProvider, region, "")
+	if pulsarClusterFromConfig == "" {
+		if cloudProvider == "" || region == "" {
+			return diag.Errorf("failed to configure resource, pulsar_cluster or cloud_provider and region must be set")
+		}
+	}
+	pulsarCluster := getPulsarCluster(pulsarClusterFromConfig, cloudProvider, region, "")
 
 	orgResp, err := astraClient.GetCurrentOrganization(ctx)
 	if err != nil {

internal/provider/util_streaming.go

@@ -171,6 +171,9 @@ func getPulsarCluster(clusterName, cloudProvider, region, suffix string) string
 	}
 	// In most astra APIs there are dashes in region names depending on the cloud provider, this seems not to be the case for streaming
 	normalizedRegion := strings.ReplaceAll(region, "-", "")
+	if cloudProvider == "" || normalizedRegion == "" {
+		return ""
+	}
 	return strings.ToLower(fmt.Sprintf("pulsar-%s-%s%s", cloudProvider, normalizedRegion, suffix))
 }