From 46e2739cd4bd15d10da8e94a6b69135e5840170c Mon Sep 17 00:00:00 2001
From: david-streamlio <35466513+david-streamlio@users.noreply.github.com>
Date: Tue, 25 Aug 2026 11:42:32 -0700
Subject: [PATCH 1/2] feat(cluster): expose broker autoscaling on
streamnative_pulsar_cluster
### Motivation
The PulsarCluster API has carried Spec.Broker.AutoScalingPolicy for as long as
the pinned cloud-api-server has, and the control plane acts on it: broker.go
translates it onto the backend cluster and, once set, stops passing a replica
count so the HPA owns the broker count.
Terraform could not reach any of it. The resource exposed only broker_replicas,
a fixed count, so a cluster managed through this provider could not be
autoscaled at all.
### Modifications
Add an optional broker_auto_scaling_policy block with min_replicas and
max_replicas, expanded onto Spec.Broker.AutoScalingPolicy on create and update
and read back on refresh.
min_replicas is Computed. The control plane picks a floor when one is not given,
and an omitted value arrives from the SDK as 0; sending that literally would ask
for a floor of no brokers. It is left nil instead, and a server-chosen value does
not read as drift.
Two failure modes are refused rather than applied silently:
- **Serverless instances.** The admission plugin pins the policy for serverless
clusters and rejects later changes to it, so a configured block could never be
honoured. Create and update refuse it outright.
- **Organizations without the feature.** Admission returns Forbidden when the
autoscaling feature gate is off. Following the maintenance_window precedent,
create and update send a dry-run first and fail when the policy that comes back
is not the one that went in, so the mismatch surfaces as an error instead of as
brokers that never scale.
broker_replicas gains no diff suppression on purpose: the resource never sets it
during read, so there is no drift to suppress. That it stops taking effect once a
policy is set is documented on the field instead.
Bookkeeper autoscaling is deliberately out of scope, and broker.scheduledScaling
is newer than the pinned cloud-api-server and would need a dependency bump.
### Verifying this change
Ten unit tests covering expand, flatten, equality, and the accepted/dropped/
overridden validation paths, including the server-defaulted min_replicas case and
the serverless override.
go build, go vet, gofmt and the full unit suite pass; docs regenerated with
tfplugindocs.
---
cloud/provider.go | 25 ++---
cloud/pulsar_cluster_state_test.go | 112 ++++++++++++++++++++
cloud/resource_pulsar_cluster.go | 165 +++++++++++++++++++++++++++++
cloud/validate_helpers.go | 12 +++
docs/resources/pulsar_cluster.md | 13 +++
5 files changed, 315 insertions(+), 12 deletions(-)
diff --git a/cloud/provider.go b/cloud/provider.go
index 1c4070f..ba05ae9 100644
--- a/cloud/provider.go
+++ b/cloud/provider.go
@@ -81,18 +81,19 @@ func init() {
"instance_engine": "The streamnative cloud instance engine, supporting 'ursa' and 'classic', default 'classic'",
"location": "The location of the pulsar cluster, " +
"supported location https://docs.streamnative.io/docs/cluster#cluster-location",
- "release_channel": "The release channel of the pulsar cluster subscribe to, it must to be lts or rapid, default rapid",
- "bookie_replicas": "The number of bookie replicas",
- "broker_replicas": "The number of broker replicas",
- "compute_unit_per_broker": "compute unit per broker, 1 compute unit is 2 cpu and 8gb memory",
- "storage_unit_per_bookie": "storage unit per bookie, 1 storage unit is 2 cpu and 8gb memory",
- "cluster_ready": "Pulsar cluster is ready, it will be set to 'True' after the cluster is ready",
- "instance_ready": "Pulsar instance is ready, it will be set to 'True' after the instance is ready",
- "websocket_enabled": "Whether the websocket is enabled",
- "function_enabled": "Whether the function is enabled",
- "transaction_enabled": "Whether the transaction is enabled",
- "kafka": "Controls the kafka protocol config of pulsar cluster",
- "mqtt": "Controls the mqtt protocol config of pulsar cluster",
+ "release_channel": "The release channel of the pulsar cluster subscribe to, it must to be lts or rapid, default rapid",
+ "bookie_replicas": "The number of bookie replicas",
+ "broker_replicas": "The number of broker replicas",
+ "broker_auto_scaling_policy": "Autoscaling bounds for brokers. When set, the number of brokers is managed automatically between min_replicas and max_replicas and broker_replicas no longer takes effect. Requires the autoscaling feature to be enabled for the organization, and is not supported on serverless instances.",
+ "compute_unit_per_broker": "compute unit per broker, 1 compute unit is 2 cpu and 8gb memory",
+ "storage_unit_per_bookie": "storage unit per bookie, 1 storage unit is 2 cpu and 8gb memory",
+ "cluster_ready": "Pulsar cluster is ready, it will be set to 'True' after the cluster is ready",
+ "instance_ready": "Pulsar instance is ready, it will be set to 'True' after the instance is ready",
+ "websocket_enabled": "Whether the websocket is enabled",
+ "function_enabled": "Whether the function is enabled",
+ "transaction_enabled": "Whether the transaction is enabled",
+ "kafka": "Controls the kafka protocol config of pulsar cluster",
+ "mqtt": "Controls the mqtt protocol config of pulsar cluster",
"categories": "Controls the audit log categories config of pulsar cluster, supported categories: " +
"\"Management\", \"Describe\", \"Produce\", \"Consume\"",
"lakehouse_type": "The type of the lakehouse",
diff --git a/cloud/pulsar_cluster_state_test.go b/cloud/pulsar_cluster_state_test.go
index f844368..685f62a 100644
--- a/cloud/pulsar_cluster_state_test.go
+++ b/cloud/pulsar_cluster_state_test.go
@@ -295,3 +295,115 @@ func TestMaintenanceWindowEqualWhenDifferent(t *testing.T) {
assert.False(t, maintenanceWindowEqual(expected, actual))
}
+
+func TestExpandBrokerAutoScalingPolicy(t *testing.T) {
+ policy := expandBrokerAutoScalingPolicy([]interface{}{
+ map[string]interface{}{
+ "min_replicas": 2,
+ "max_replicas": 6,
+ },
+ })
+
+ assert.NotNil(t, policy)
+ assert.Equal(t, int32(6), policy.MaxReplicas)
+ assert.NotNil(t, policy.MinReplicas)
+ assert.Equal(t, int32(2), *policy.MinReplicas)
+}
+
+// min_replicas is Computed, so an omitted value arrives as 0. Sending that literally would ask the
+// control plane for a floor of no brokers, so it has to stay nil and be filled in server side.
+func TestExpandBrokerAutoScalingPolicyOmittedMinReplicas(t *testing.T) {
+ policy := expandBrokerAutoScalingPolicy([]interface{}{
+ map[string]interface{}{
+ "min_replicas": 0,
+ "max_replicas": 4,
+ },
+ })
+
+ assert.NotNil(t, policy)
+ assert.Equal(t, int32(4), policy.MaxReplicas)
+ assert.Nil(t, policy.MinReplicas)
+}
+
+func TestExpandBrokerAutoScalingPolicyEmpty(t *testing.T) {
+ assert.Nil(t, expandBrokerAutoScalingPolicy(nil))
+ assert.Nil(t, expandBrokerAutoScalingPolicy([]interface{}{}))
+ assert.Nil(t, expandBrokerAutoScalingPolicy([]interface{}{nil}))
+}
+
+func TestFlattenBrokerAutoScalingPolicy(t *testing.T) {
+ minReplicas := int32(3)
+ flattened := flattenBrokerAutoScalingPolicy(&cloudv1alpha1.AutoScalingPolicy{
+ MinReplicas: &minReplicas,
+ MaxReplicas: 9,
+ })
+
+ assert.Len(t, flattened, 1)
+ assert.Equal(t, map[string]interface{}{"min_replicas": 3, "max_replicas": 9}, flattened[0])
+ assert.Equal(t, []interface{}{}, flattenBrokerAutoScalingPolicy(nil))
+}
+
+func TestBrokerAutoScalingPolicyEqual(t *testing.T) {
+ minReplicas := int32(2)
+ expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+ actualMin := int32(2)
+ actual := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &actualMin, MaxReplicas: 6}
+
+ assert.True(t, brokerAutoScalingPolicyEqual(expected, actual))
+ assert.True(t, brokerAutoScalingPolicyEqual(nil, nil))
+ assert.False(t, brokerAutoScalingPolicyEqual(expected, nil))
+ assert.False(t, brokerAutoScalingPolicyEqual(nil, actual))
+}
+
+func TestBrokerAutoScalingPolicyEqualWhenDifferent(t *testing.T) {
+ minReplicas := int32(2)
+ expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+
+ differentMax := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 7}
+ assert.False(t, brokerAutoScalingPolicyEqual(expected, differentMax))
+
+ otherMin := int32(3)
+ differentMin := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &otherMin, MaxReplicas: 6}
+ assert.False(t, brokerAutoScalingPolicyEqual(expected, differentMin))
+
+ droppedMin := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+ assert.False(t, brokerAutoScalingPolicyEqual(expected, droppedMin))
+}
+
+// A min_replicas the user did not ask for is the control plane's to choose, so a server-filled value
+// must not read as drift.
+func TestBrokerAutoScalingPolicyEqualWhenMinReplicasDefaulted(t *testing.T) {
+ expected := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+ serverFilled := int32(2)
+ actual := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &serverFilled, MaxReplicas: 6}
+
+ assert.True(t, brokerAutoScalingPolicyEqual(expected, actual))
+}
+
+func TestValidateBrokerAutoScalingPolicyAccepted(t *testing.T) {
+ minReplicas := int32(2)
+ expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+
+ assert.NoError(t, validateBrokerAutoScalingPolicyAccepted(expected, expected, "CREATE"))
+}
+
+func TestValidateBrokerAutoScalingPolicyAcceptedWhenDropped(t *testing.T) {
+ expected := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+
+ err := validateBrokerAutoScalingPolicyAccepted(expected, nil, "CREATE")
+ assert.EqualError(t, err,
+ "ERROR_CREATE_PULSAR_CLUSTER: broker_auto_scaling_policy is not enabled for this organization")
+}
+
+// The control plane pins serverless clusters to its own policy; catching that as a mismatch is what
+// stops a silently overridden configuration from being reported as applied.
+func TestValidateBrokerAutoScalingPolicyAcceptedWhenOverridden(t *testing.T) {
+ requestedMin := int32(1)
+ expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &requestedMin, MaxReplicas: 10}
+ pinnedMin := int32(2)
+ actual := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &pinnedMin, MaxReplicas: 3}
+
+ err := validateBrokerAutoScalingPolicyAccepted(expected, actual, "UPDATE")
+ assert.EqualError(t, err,
+ "ERROR_UPDATE_PULSAR_CLUSTER: broker_auto_scaling_policy is not enabled for this organization")
+}
diff --git a/cloud/resource_pulsar_cluster.go b/cloud/resource_pulsar_cluster.go
index ef3a353..fab8cd5 100644
--- a/cloud/resource_pulsar_cluster.go
+++ b/cloud/resource_pulsar_cluster.go
@@ -381,6 +381,29 @@ func resourcePulsarCluster() *schema.Resource {
Computed: true,
Description: descriptions["iam_policy"],
},
+ "broker_auto_scaling_policy": {
+ Type: schema.TypeList,
+ Optional: true,
+ MaxItems: 1,
+ Description: descriptions["broker_auto_scaling_policy"],
+ Elem: &schema.Resource{
+ Schema: map[string]*schema.Schema{
+ "min_replicas": {
+ Type: schema.TypeInt,
+ Optional: true,
+ Computed: true,
+ Description: "The minimum number of brokers to scale down to. Defaults to the value chosen by the control plane when omitted.",
+ ValidateFunc: validateAutoScalingReplicas,
+ },
+ "max_replicas": {
+ Type: schema.TypeInt,
+ Required: true,
+ Description: "The maximum number of brokers to scale up to.",
+ ValidateFunc: validateAutoScalingReplicas,
+ },
+ },
+ },
+ },
"maintenance_window": {
Type: schema.TypeList,
Optional: true,
@@ -516,6 +539,12 @@ func resourcePulsarClusterCreate(ctx context.Context, d *schema.ResourceData, me
return diag.FromErr(fmt.Errorf("ERROR_CREATE_PULSAR_CLUSTER: " +
"broker_replicas must be 2 for serverless instance"))
}
+ if hasBrokerAutoScalingPolicyConfigured(d) {
+ // The control plane pins the policy for serverless clusters and refuses later changes to
+ // it, so a configured block here could never be honoured.
+ return diag.FromErr(fmt.Errorf("ERROR_CREATE_PULSAR_CLUSTER: " +
+ "broker_auto_scaling_policy is not supported for serverless instance"))
+ }
pulsarCluster.Annotations = map[string]string{
"cloud.streamnative.io/type": "serverless",
}
@@ -566,6 +595,9 @@ func resourcePulsarClusterCreate(ctx context.Context, d *schema.ResourceData, me
}
}
pulsarCluster.Spec.MaintenanceWindow = expandMaintenanceWindow(ctx, d.Get("maintenance_window").([]interface{}))
+ if autoScalingPolicy := expandBrokerAutoScalingPolicy(d.Get("broker_auto_scaling_policy").([]interface{})); autoScalingPolicy != nil {
+ pulsarCluster.Spec.Broker.AutoScalingPolicy = autoScalingPolicy
+ }
if pulsarInstance.Spec.Type != cloudv1alpha1.PulsarInstanceTypeServerless && !pulsarInstance.IsUsingUrsaEngine() {
getPulsarClusterChanged(ctx, pulsarCluster, d)
}
@@ -648,6 +680,9 @@ func resourcePulsarClusterCreate(ctx context.Context, d *schema.ResourceData, me
if diagErr := ensureMaintenanceWindowAcceptedOnDryRun(ctx, clientSet, namespace, pulsarCluster, "CREATE"); diagErr != nil {
return diagErr
}
+ if diagErr := ensureBrokerAutoScalingPolicyAcceptedOnDryRun(ctx, clientSet, namespace, pulsarCluster, "CREATE"); diagErr != nil {
+ return diagErr
+ }
pc, err := clientSet.CloudV1alpha1().PulsarClusters(namespace).Create(ctx, pulsarCluster, metav1.CreateOptions{
FieldManager: "terraform-create",
@@ -834,6 +869,14 @@ func resourcePulsarClusterRead(ctx context.Context, d *schema.ResourceData, meta
} else {
_ = d.Set("maintenance_window", []interface{}{})
}
+ if pulsarCluster.Spec.Broker.AutoScalingPolicy != nil {
+ err = d.Set("broker_auto_scaling_policy", flattenBrokerAutoScalingPolicy(pulsarCluster.Spec.Broker.AutoScalingPolicy))
+ if err != nil {
+ return diag.FromErr(fmt.Errorf("ERROR_READ_PULSAR_CLUSTER_BROKER_AUTO_SCALING_POLICY: %w", err))
+ }
+ } else {
+ _ = d.Set("broker_auto_scaling_policy", []interface{}{})
+ }
if pulsarInstance.Spec.Type != cloudv1alpha1.PulsarInstanceTypeServerless && !pulsarCluster.IsUsingUrsaEngine() {
bookkeeperImage := strings.Split(pulsarCluster.Spec.BookKeeper.Image, ":")
if len(bookkeeperImage) > 1 {
@@ -1065,6 +1108,15 @@ func resourcePulsarClusterUpdate(ctx context.Context, d *schema.ResourceData, me
pulsarCluster.Spec.MaintenanceWindow = expandMaintenanceWindow(ctx, d.Get("maintenance_window").([]interface{}))
changed = true
}
+ if d.HasChange("broker_auto_scaling_policy") {
+ if d.Get("type") == string(cloudv1alpha1.PulsarInstanceTypeServerless) {
+ return diag.FromErr(fmt.Errorf("ERROR_UPDATE_PULSAR_CLUSTER: " +
+ "broker_auto_scaling_policy is not supported for serverless instance"))
+ }
+ pulsarCluster.Spec.Broker.AutoScalingPolicy = expandBrokerAutoScalingPolicy(
+ d.Get("broker_auto_scaling_policy").([]interface{}))
+ changed = true
+ }
if displayNameChanged {
displayName := d.Get("display_name").(string)
pulsarCluster.Spec.DisplayName = displayName
@@ -1169,6 +1221,11 @@ func resourcePulsarClusterUpdate(ctx context.Context, d *schema.ResourceData, me
return diagErr
}
}
+ if d.HasChange("broker_auto_scaling_policy") && hasBrokerAutoScalingPolicyConfigured(d) {
+ if diagErr := ensureBrokerAutoScalingPolicyAcceptedOnDryRun(ctx, clientSet, namespace, pulsarCluster, "UPDATE"); diagErr != nil {
+ return diagErr
+ }
+ }
_, err = clientSet.CloudV1alpha1().PulsarClusters(namespace).Update(ctx, pulsarCluster, metav1.UpdateOptions{
FieldManager: "terraform-update",
})
@@ -1373,6 +1430,114 @@ func getPulsarClusterChanged(ctx context.Context, pulsarCluster *cloudv1alpha1.P
return changed
}
+func hasBrokerAutoScalingPolicyConfigured(d *schema.ResourceData) bool {
+ return len(d.Get("broker_auto_scaling_policy").([]interface{})) > 0
+}
+
+// expandBrokerAutoScalingPolicy builds the API policy from configuration. min_replicas is Computed,
+// so an omitted value reads back as 0; that is left nil for the control plane to fill in rather than
+// sent as a literal zero, which would ask for a scale-to-nothing floor.
+func expandBrokerAutoScalingPolicy(policy []interface{}) *cloudv1alpha1.AutoScalingPolicy {
+ if len(policy) == 0 || policy[0] == nil {
+ return nil
+ }
+ policyMap, ok := policy[0].(map[string]interface{})
+ if !ok {
+ return nil
+ }
+ result := &cloudv1alpha1.AutoScalingPolicy{}
+ if maxReplicas, ok := policyMap["max_replicas"].(int); ok {
+ result.MaxReplicas = int32(maxReplicas)
+ }
+ if minReplicas, ok := policyMap["min_replicas"].(int); ok && minReplicas > 0 {
+ value := int32(minReplicas)
+ result.MinReplicas = &value
+ }
+ return result
+}
+
+func flattenBrokerAutoScalingPolicy(policy *cloudv1alpha1.AutoScalingPolicy) []interface{} {
+ if policy == nil {
+ return []interface{}{}
+ }
+ result := map[string]interface{}{
+ "max_replicas": int(policy.MaxReplicas),
+ }
+ if policy.MinReplicas != nil {
+ result["min_replicas"] = int(*policy.MinReplicas)
+ }
+ return []interface{}{result}
+}
+
+func brokerAutoScalingPolicyEqual(expected *cloudv1alpha1.AutoScalingPolicy, actual *cloudv1alpha1.AutoScalingPolicy) bool {
+ if expected == nil || actual == nil {
+ return expected == nil && actual == nil
+ }
+ if expected.MaxReplicas != actual.MaxReplicas {
+ return false
+ }
+ // An omitted min_replicas is filled in by the control plane, so only a value the user asked for
+ // has to survive the round trip.
+ if expected.MinReplicas == nil {
+ return true
+ }
+ return actual.MinReplicas != nil && *expected.MinReplicas == *actual.MinReplicas
+}
+
+func validateBrokerAutoScalingPolicyAccepted(
+ expected *cloudv1alpha1.AutoScalingPolicy,
+ actual *cloudv1alpha1.AutoScalingPolicy,
+ operation string,
+) error {
+ if brokerAutoScalingPolicyEqual(expected, actual) {
+ return nil
+ }
+ return fmt.Errorf(
+ "ERROR_%s_PULSAR_CLUSTER: broker_auto_scaling_policy is not enabled for this organization",
+ operation,
+ )
+}
+
+// ensureBrokerAutoScalingPolicyAcceptedOnDryRun asks the control plane to evaluate the cluster without
+// persisting it, and fails when the policy that comes back is not the one that went in. The admission
+// plugin refuses the field outright when the organization lacks the autoscaling feature, so the dry run
+// surfaces that as a create/update error rather than as silently unscaled brokers.
+func ensureBrokerAutoScalingPolicyAcceptedOnDryRun(
+ ctx context.Context,
+ clientSet *cloudclient.Clientset,
+ namespace string,
+ pulsarCluster *cloudv1alpha1.PulsarCluster,
+ operation string,
+) diag.Diagnostics {
+ if pulsarCluster.Spec.Broker.AutoScalingPolicy == nil {
+ return nil
+ }
+ var (
+ preview *cloudv1alpha1.PulsarCluster
+ err error
+ )
+ options := metav1.UpdateOptions{
+ FieldManager: fmt.Sprintf("terraform-%s-broker-auto-scaling-validation", strings.ToLower(operation)),
+ DryRun: []string{metav1.DryRunAll},
+ }
+ if operation == "CREATE" {
+ preview, err = clientSet.CloudV1alpha1().PulsarClusters(namespace).Create(ctx, pulsarCluster, metav1.CreateOptions{
+ FieldManager: options.FieldManager,
+ DryRun: options.DryRun,
+ })
+ } else {
+ preview, err = clientSet.CloudV1alpha1().PulsarClusters(namespace).Update(ctx, pulsarCluster, options)
+ }
+ if err != nil {
+ return diag.FromErr(fmt.Errorf("ERROR_VALIDATE_BROKER_AUTO_SCALING_POLICY_ON_%s_PULSAR_CLUSTER: %w", operation, err))
+ }
+ if err := validateBrokerAutoScalingPolicyAccepted(
+ pulsarCluster.Spec.Broker.AutoScalingPolicy, preview.Spec.Broker.AutoScalingPolicy, operation); err != nil {
+ return diag.FromErr(err)
+ }
+ return nil
+}
+
func hasMaintenanceWindowConfigured(d *schema.ResourceData) bool {
return len(d.Get("maintenance_window").([]interface{})) > 0
}
diff --git a/cloud/validate_helpers.go b/cloud/validate_helpers.go
index 5f57391..ec85b59 100644
--- a/cloud/validate_helpers.go
+++ b/cloud/validate_helpers.go
@@ -325,3 +325,15 @@ var validRegions = []string{
"uaecentral",
"brazilsoutheas",
}
+
+// validateAutoScalingReplicas bounds an autoscaling replica count to the same range the fixed
+// broker_replicas count allows, so a policy cannot ask for a cluster size that could not be
+// requested directly.
+func validateAutoScalingReplicas(val interface{}, key string) (warns []string, errs []error) {
+ v := val.(int)
+ if v < 1 || v > 15 {
+ errs = append(errs, fmt.Errorf(
+ "%q should be greater than or equal to 1 and less than or equal to 15, got: %d", key, v))
+ }
+ return
+}
diff --git a/docs/resources/pulsar_cluster.md b/docs/resources/pulsar_cluster.md
index b90e143..38a13b1 100644
--- a/docs/resources/pulsar_cluster.md
+++ b/docs/resources/pulsar_cluster.md
@@ -24,6 +24,7 @@ description: |-
- `apply_lakehouse_to_all_topics` (Boolean) Whether to apply lakehouse storage to all topics in the cluster
- `bookie_replicas` (Number) The number of bookie replicas
+- `broker_auto_scaling_policy` (Block List, Max: 1) Autoscaling bounds for brokers. When set, the number of brokers is managed automatically between min_replicas and max_replicas and broker_replicas no longer takes effect. Requires the autoscaling feature to be enabled for the organization, and is not supported on serverless instances. (see [below for nested schema](#nestedblock--broker_auto_scaling_policy))
- `broker_replicas` (Number) The number of broker replicas
- `catalog` (String) The name of the catalog to use for this pulsar cluster
- `compute_unit` (Number, Deprecated) compute unit per broker, 1 compute unit is 2 cpu and 8gb memory
@@ -61,6 +62,18 @@ description: |-
- `websocket_service_url` (String) If you want to connect to the pulsar cluster using the websocket protocol, use this websocket service url.
- `websocket_service_urls` (List of String) If you want to connect to the pulsar cluster using the websocket protocol, use this websocket service url. There'll be multiple service urls if the cluster attached with multiple gateways
+
+### Nested Schema for `broker_auto_scaling_policy`
+
+Required:
+
+- `max_replicas` (Number) The maximum number of brokers to scale up to.
+
+Optional:
+
+- `min_replicas` (Number) The minimum number of brokers to scale down to. Defaults to the value chosen by the control plane when omitted.
+
+
### Nested Schema for `config`
From 90d417b3c96ad621c16914c59209e8c718c95422 Mon Sep 17 00:00:00 2001
From: david-streamlio <35466513+david-streamlio@users.noreply.github.com>
Date: Wed, 26 Aug 2026 08:54:00 -0700
Subject: [PATCH 2/2] fix(cluster): address broker autoscaling review on
min_replicas, refresh, and bounds
Motivation:
Review of #159 surfaced three correctness issues in broker_auto_scaling_policy.
Modifications:
- Make min_replicas Required. sn-operator's GetAutoScalingPolicyWithDefault
returns a nil policy whenever MinReplicas is nil, so it builds no
HorizontalPodAutoscaler at all; a configuration carrying only max_replicas
previously passed both this provider and the cloud API while leaving
autoscaling off. expandBrokerAutoScalingPolicy now always sends the minimum,
and the dry-run comparison no longer treats an absent minimum as accepted.
- Stop adopting an unmanaged policy during refresh. The block is Optional with
no Computed, so a policy Terraform never wrote made an absent configuration
plan broker_auto_scaling_policy.# from 1 to 0: on serverless, admission
injects a policy and the update guard then rejected every apply of that
removal; on a dedicated cluster whose policy was set out of band, the same
plan silently disabled autoscaling. brokerAutoScalingPolicyForState now
surfaces only a policy already under management and skips serverless.
- Validate the bounds as a pair in CustomizeDiff. The scalar validators bound
each field independently and cloud-api admission has no cross-field check, so
min_replicas > max_replicas surfaced as a Kubernetes reconcile failure well
after apply reported success; it is now a plan-time diagnostic.
Verification:
- go build ./... , go vet ./cloud/ , go test ./cloud/ all pass
- Added unit coverage for the schema requirement, the refresh decision, the
bounds helper, and plan-level regressions via Resource.SimpleDiff for both
the unmanaged-policy and inverted-range cases
- Regenerated docs with tfplugindocs v0.19.3
---
cloud/pulsar_cluster_state_test.go | 112 ++++++++++++++++++++++++++---
cloud/resource_pulsar_cluster.go | 79 +++++++++++++++-----
docs/resources/pulsar_cluster.md | 5 +-
3 files changed, 163 insertions(+), 33 deletions(-)
diff --git a/cloud/pulsar_cluster_state_test.go b/cloud/pulsar_cluster_state_test.go
index 685f62a..a71d50c 100644
--- a/cloud/pulsar_cluster_state_test.go
+++ b/cloud/pulsar_cluster_state_test.go
@@ -20,6 +20,7 @@ import (
"time"
"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+ "github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
cloudv1alpha1 "github.com/streamnative/cloud-api-server/pkg/apis/cloud/v1alpha1"
"github.com/stretchr/testify/assert"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -310,19 +311,31 @@ func TestExpandBrokerAutoScalingPolicy(t *testing.T) {
assert.Equal(t, int32(2), *policy.MinReplicas)
}
-// min_replicas is Computed, so an omitted value arrives as 0. Sending that literally would ask the
-// control plane for a floor of no brokers, so it has to stay nil and be filled in server side.
-func TestExpandBrokerAutoScalingPolicyOmittedMinReplicas(t *testing.T) {
+// min_replicas is Required, so the schema rejects an omitted value before expand is reached. What
+// expand must never do is drop a minimum it was given: sn-operator's GetAutoScalingPolicyWithDefault
+// returns a nil policy whenever MinReplicas is nil, which builds no HorizontalPodAutoscaler and
+// leaves autoscaling silently off.
+func TestExpandBrokerAutoScalingPolicyAlwaysSendsMinReplicas(t *testing.T) {
policy := expandBrokerAutoScalingPolicy([]interface{}{
map[string]interface{}{
- "min_replicas": 0,
+ "min_replicas": 1,
"max_replicas": 4,
},
})
assert.NotNil(t, policy)
assert.Equal(t, int32(4), policy.MaxReplicas)
- assert.Nil(t, policy.MinReplicas)
+ assert.NotNil(t, policy.MinReplicas)
+ assert.Equal(t, int32(1), *policy.MinReplicas)
+}
+
+func TestBrokerAutoScalingPolicySchemaRequiresMinReplicas(t *testing.T) {
+ elem := resourcePulsarCluster().Schema["broker_auto_scaling_policy"].Elem.(*schema.Resource)
+
+ assert.True(t, elem.Schema["min_replicas"].Required,
+ "min_replicas must be Required; an absent minimum disables the downstream HPA")
+ assert.False(t, elem.Schema["min_replicas"].Computed,
+ "min_replicas must not be Computed; a server-filled value would mask the absent minimum")
}
func TestExpandBrokerAutoScalingPolicyEmpty(t *testing.T) {
@@ -372,12 +385,20 @@ func TestBrokerAutoScalingPolicyEqualWhenDifferent(t *testing.T) {
// A min_replicas the user did not ask for is the control plane's to choose, so a server-filled value
// must not read as drift.
-func TestBrokerAutoScalingPolicyEqualWhenMinReplicasDefaulted(t *testing.T) {
- expected := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+// The dry-run comparison used to treat a nil expected minimum as "whatever the control plane fills
+// in", which hid the case this PR now prevents: a request carrying no minimum builds no HPA at all.
+// min_replicas is Required, so a nil on either side is a real mismatch rather than a defaulted value.
+func TestBrokerAutoScalingPolicyEqualRejectsMissingMinReplicas(t *testing.T) {
serverFilled := int32(2)
- actual := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &serverFilled, MaxReplicas: 6}
-
- assert.True(t, brokerAutoScalingPolicyEqual(expected, actual))
+ withMin := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &serverFilled, MaxReplicas: 6}
+ withoutMin := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+
+ assert.False(t, brokerAutoScalingPolicyEqual(withoutMin, withMin),
+ "a minimum the request never asked for is not an accepted policy")
+ assert.False(t, brokerAutoScalingPolicyEqual(withMin, withoutMin),
+ "a dropped minimum means the policy was not accepted as sent")
+ assert.True(t, brokerAutoScalingPolicyEqual(withoutMin, withoutMin),
+ "two policies without a minimum still compare equal")
}
func TestValidateBrokerAutoScalingPolicyAccepted(t *testing.T) {
@@ -407,3 +428,74 @@ func TestValidateBrokerAutoScalingPolicyAcceptedWhenOverridden(t *testing.T) {
assert.EqualError(t, err,
"ERROR_UPDATE_PULSAR_CLUSTER: broker_auto_scaling_policy is not enabled for this organization")
}
+
+// An unmanaged policy must never reach state. The block is Optional with no Computed, so a policy
+// Terraform did not write would make an absent configuration plan broker_auto_scaling_policy.# from
+// 1 to 0 -- rejected by the update guard on serverless, and a silent disable on dedicated.
+func TestBrokerAutoScalingPolicyForState(t *testing.T) {
+ minReplicas := int32(2)
+ policy := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+ managed := []interface{}{map[string]interface{}{"min_replicas": 2, "max_replicas": 6}}
+
+ assert.Equal(t, managed, brokerAutoScalingPolicyForState(policy, true, false),
+ "a policy Terraform manages on a dedicated cluster is surfaced")
+ assert.Equal(t, []interface{}{}, brokerAutoScalingPolicyForState(policy, false, false),
+ "a policy set out of band is not adopted into state")
+ assert.Equal(t, []interface{}{}, brokerAutoScalingPolicyForState(policy, true, true),
+ "the serverless admission-injected policy is never surfaced")
+ assert.Equal(t, []interface{}{}, brokerAutoScalingPolicyForState(nil, true, false),
+ "an absent policy clears state so removal drift is still detected")
+}
+
+// Plan-level regression for the same issue: once refresh leaves an unmanaged policy out of state, a
+// configuration without the block must produce no broker_auto_scaling_policy change at all.
+func TestBrokerAutoScalingPolicyUnmanagedProducesNoPlan(t *testing.T) {
+ state := &terraform.InstanceState{
+ ID: "test-cluster",
+ Attributes: map[string]string{"id": "test-cluster"},
+ }
+
+ diff, err := resourcePulsarCluster().SimpleDiff(
+ context.Background(), state, terraform.NewResourceConfigRaw(map[string]interface{}{}), nil)
+ assert.NoError(t, err)
+ if diff != nil {
+ _, planned := diff.Attributes["broker_auto_scaling_policy.#"]
+ assert.False(t, planned,
+ "an unmanaged policy absent from state must not plan a broker_auto_scaling_policy change")
+ }
+}
+
+func TestValidateBrokerAutoScalingPolicyBounds(t *testing.T) {
+ boundsFor := func(minReplicas, maxReplicas int) []interface{} {
+ return []interface{}{
+ map[string]interface{}{"min_replicas": minReplicas, "max_replicas": maxReplicas},
+ }
+ }
+
+ err := validateBrokerAutoScalingPolicyBounds(boundsFor(10, 5))
+ assert.Error(t, err, "an inverted range must fail at plan time, not at Kubernetes reconcile")
+ assert.Contains(t, err.Error(), "min_replicas (10) must be less than or equal to max_replicas (5)")
+
+ assert.NoError(t, validateBrokerAutoScalingPolicyBounds(boundsFor(3, 3)), "an equal range is valid")
+ assert.NoError(t, validateBrokerAutoScalingPolicyBounds(boundsFor(2, 6)), "an ascending range is valid")
+ assert.NoError(t, validateBrokerAutoScalingPolicyBounds(nil), "an absent block carries no bounds")
+ assert.NoError(t, validateBrokerAutoScalingPolicyBounds([]interface{}{nil}), "a nil entry is ignored")
+}
+
+// The inverted range must surface through the resource's CustomizeDiff, not just the helper.
+func TestBrokerAutoScalingPolicyBoundsRejectedAtPlan(t *testing.T) {
+ state := &terraform.InstanceState{
+ ID: "test-cluster",
+ Attributes: map[string]string{"id": "test-cluster"},
+ }
+
+ _, err := resourcePulsarCluster().SimpleDiff(context.Background(), state,
+ terraform.NewResourceConfigRaw(map[string]interface{}{
+ "broker_auto_scaling_policy": []interface{}{
+ map[string]interface{}{"min_replicas": 10, "max_replicas": 5},
+ },
+ }), nil)
+
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "min_replicas (10) must be less than or equal to max_replicas (5)")
+}
diff --git a/cloud/resource_pulsar_cluster.go b/cloud/resource_pulsar_cluster.go
index fab8cd5..b80b688 100644
--- a/cloud/resource_pulsar_cluster.go
+++ b/cloud/resource_pulsar_cluster.go
@@ -42,6 +42,12 @@ func resourcePulsarCluster() *schema.Resource {
UpdateContext: resourcePulsarClusterUpdate,
DeleteContext: resourcePulsarClusterDelete,
CustomizeDiff: func(ctx context.Context, diff *schema.ResourceDiff, i interface{}) error {
+ // Runs ahead of the early returns below so the bounds hold on create and update alike.
+ if policy, ok := diff.Get("broker_auto_scaling_policy").([]interface{}); ok {
+ if err := validateBrokerAutoScalingPolicyBounds(policy); err != nil {
+ return err
+ }
+ }
oldOrg, _ := diff.GetChange("organization")
oldName, newName := diff.GetChange("name")
if oldOrg.(string) == "" && oldName.(string) == "" {
@@ -390,9 +396,8 @@ func resourcePulsarCluster() *schema.Resource {
Schema: map[string]*schema.Schema{
"min_replicas": {
Type: schema.TypeInt,
- Optional: true,
- Computed: true,
- Description: "The minimum number of brokers to scale down to. Defaults to the value chosen by the control plane when omitted.",
+ Required: true,
+ Description: "The minimum number of brokers to scale down to. Must be set explicitly: the operator builds no HorizontalPodAutoscaler at all when the minimum is absent, so an omitted value would leave autoscaling silently off.",
ValidateFunc: validateAutoScalingReplicas,
},
"max_replicas": {
@@ -869,13 +874,13 @@ func resourcePulsarClusterRead(ctx context.Context, d *schema.ResourceData, meta
} else {
_ = d.Set("maintenance_window", []interface{}{})
}
- if pulsarCluster.Spec.Broker.AutoScalingPolicy != nil {
- err = d.Set("broker_auto_scaling_policy", flattenBrokerAutoScalingPolicy(pulsarCluster.Spec.Broker.AutoScalingPolicy))
- if err != nil {
- return diag.FromErr(fmt.Errorf("ERROR_READ_PULSAR_CLUSTER_BROKER_AUTO_SCALING_POLICY: %w", err))
- }
- } else {
- _ = d.Set("broker_auto_scaling_policy", []interface{}{})
+ err = d.Set("broker_auto_scaling_policy", brokerAutoScalingPolicyForState(
+ pulsarCluster.Spec.Broker.AutoScalingPolicy,
+ hasBrokerAutoScalingPolicyConfigured(d),
+ pulsarInstance.Spec.Type == cloudv1alpha1.PulsarInstanceTypeServerless,
+ ))
+ if err != nil {
+ return diag.FromErr(fmt.Errorf("ERROR_READ_PULSAR_CLUSTER_BROKER_AUTO_SCALING_POLICY: %w", err))
}
if pulsarInstance.Spec.Type != cloudv1alpha1.PulsarInstanceTypeServerless && !pulsarCluster.IsUsingUrsaEngine() {
bookkeeperImage := strings.Split(pulsarCluster.Spec.BookKeeper.Image, ":")
@@ -1434,9 +1439,9 @@ func hasBrokerAutoScalingPolicyConfigured(d *schema.ResourceData) bool {
return len(d.Get("broker_auto_scaling_policy").([]interface{})) > 0
}
-// expandBrokerAutoScalingPolicy builds the API policy from configuration. min_replicas is Computed,
-// so an omitted value reads back as 0; that is left nil for the control plane to fill in rather than
-// sent as a literal zero, which would ask for a scale-to-nothing floor.
+// expandBrokerAutoScalingPolicy builds the API policy from configuration. min_replicas is Required
+// because sn-operator's GetAutoScalingPolicyWithDefault returns a nil policy whenever MinReplicas is
+// nil, which builds no HorizontalPodAutoscaler and leaves autoscaling silently off.
func expandBrokerAutoScalingPolicy(policy []interface{}) *cloudv1alpha1.AutoScalingPolicy {
if len(policy) == 0 || policy[0] == nil {
return nil
@@ -1449,13 +1454,51 @@ func expandBrokerAutoScalingPolicy(policy []interface{}) *cloudv1alpha1.AutoScal
if maxReplicas, ok := policyMap["max_replicas"].(int); ok {
result.MaxReplicas = int32(maxReplicas)
}
- if minReplicas, ok := policyMap["min_replicas"].(int); ok && minReplicas > 0 {
+ if minReplicas, ok := policyMap["min_replicas"].(int); ok {
value := int32(minReplicas)
result.MinReplicas = &value
}
return result
}
+// brokerAutoScalingPolicyForState decides what refresh writes into state. Only a policy Terraform
+// already manages may be surfaced: the block is Optional with no Computed, so importing an unmanaged
+// policy would make an absent configuration plan broker_auto_scaling_policy.# from 1 to 0. On a
+// serverless instance admission injects a policy the practitioner never wrote, and the update guard
+// then rejects every apply of that removal; on a dedicated cluster whose policy was set out of band,
+// the same plan would silently disable autoscaling.
+func brokerAutoScalingPolicyForState(
+ policy *cloudv1alpha1.AutoScalingPolicy, managed bool, serverless bool,
+) []interface{} {
+ if policy == nil || !managed || serverless {
+ return []interface{}{}
+ }
+ return flattenBrokerAutoScalingPolicy(policy)
+}
+
+// validateBrokerAutoScalingPolicyBounds rejects an inverted range at plan time. The scalar
+// validators only bound each field independently, and cloud-api admission performs no cross-field
+// check, so min_replicas > max_replicas would otherwise surface as a Kubernetes reconcile failure
+// long after apply reported success.
+func validateBrokerAutoScalingPolicyBounds(policy []interface{}) error {
+ if len(policy) == 0 || policy[0] == nil {
+ return nil
+ }
+ policyMap, ok := policy[0].(map[string]interface{})
+ if !ok {
+ return nil
+ }
+ minReplicas, _ := policyMap["min_replicas"].(int)
+ maxReplicas, _ := policyMap["max_replicas"].(int)
+ if minReplicas <= 0 || maxReplicas <= 0 || minReplicas <= maxReplicas {
+ return nil
+ }
+ return fmt.Errorf(
+ "ERROR_PULSAR_CLUSTER_BROKER_AUTO_SCALING_POLICY: "+
+ "min_replicas (%d) must be less than or equal to max_replicas (%d)",
+ minReplicas, maxReplicas)
+}
+
func flattenBrokerAutoScalingPolicy(policy *cloudv1alpha1.AutoScalingPolicy) []interface{} {
if policy == nil {
return []interface{}{}
@@ -1476,12 +1519,10 @@ func brokerAutoScalingPolicyEqual(expected *cloudv1alpha1.AutoScalingPolicy, act
if expected.MaxReplicas != actual.MaxReplicas {
return false
}
- // An omitted min_replicas is filled in by the control plane, so only a value the user asked for
- // has to survive the round trip.
- if expected.MinReplicas == nil {
- return true
+ if expected.MinReplicas == nil || actual.MinReplicas == nil {
+ return expected.MinReplicas == nil && actual.MinReplicas == nil
}
- return actual.MinReplicas != nil && *expected.MinReplicas == *actual.MinReplicas
+ return *expected.MinReplicas == *actual.MinReplicas
}
func validateBrokerAutoScalingPolicyAccepted(
diff --git a/docs/resources/pulsar_cluster.md b/docs/resources/pulsar_cluster.md
index 38a13b1..342b016 100644
--- a/docs/resources/pulsar_cluster.md
+++ b/docs/resources/pulsar_cluster.md
@@ -68,10 +68,7 @@ description: |-
Required:
- `max_replicas` (Number) The maximum number of brokers to scale up to.
-
-Optional:
-
-- `min_replicas` (Number) The minimum number of brokers to scale down to. Defaults to the value chosen by the control plane when omitted.
+- `min_replicas` (Number) The minimum number of brokers to scale down to. Must be set explicitly: the operator builds no HorizontalPodAutoscaler at all when the minimum is absent, so an omitted value would leave autoscaling silently off.