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..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"
@@ -295,3 +296,206 @@ 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 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": 1,
+			"max_replicas": 4,
+		},
+	})
+
+	assert.NotNil(t, policy)
+	assert.Equal(t, int32(4), policy.MaxReplicas)
+	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) {
+	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.
+// 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)
+	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) {
+	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")
+}
+
+// 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 ef3a353..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) == "" {
@@ -381,6 +387,28 @@ 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,
+							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": {
+							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 +544,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 +600,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 +685,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 +874,14 @@ func resourcePulsarClusterRead(ctx context.Context, d *schema.ResourceData, meta
 	} else {
 		_ = d.Set("maintenance_window", []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, ":")
 		if len(bookkeeperImage) > 1 {
@@ -1065,6 +1113,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 +1226,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 +1435,150 @@ 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 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
+	}
+	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 {
+		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{}{}
+	}
+	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
+	}
+	if expected.MinReplicas == nil || actual.MinReplicas == nil {
+		return expected.MinReplicas == nil && actual.MinReplicas == nil
+	}
+	return *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..342b016 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,15 @@ 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
 
+<a id="nestedblock--broker_auto_scaling_policy"></a>
+### Nested Schema for `broker_auto_scaling_policy`
+
+Required:
+
+- `max_replicas` (Number) The maximum number of brokers to scale up to.
+- `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.
+
+
 <a id="nestedblock--config"></a>
 ### Nested Schema for `config`