Support topic_auto_creation

go.mod

@@ -3,7 +3,7 @@ module github.com/streamnative/terraform-provider-pulsar
 go 1.18
 
 require (
-	github.com/apache/pulsar-client-go v0.9.1-0.20230816081803-fbee610ddcbf
+	github.com/apache/pulsar-client-go v0.12.0
 	github.com/cenkalti/backoff/v4 v4.1.2
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/terraform-plugin-log v0.4.0

go.sum

@@ -16,8 +16,8 @@ github.com/agext/levenshtein v1.2.1/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki
 github.com/agext/levenshtein v1.2.2 h1:0S/Yg6LYmFJ5stwQeRp6EeOcCbj7xiqQSdNelsXvaqE=
 github.com/agext/levenshtein v1.2.2/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558=
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
-github.com/apache/pulsar-client-go v0.9.1-0.20230816081803-fbee610ddcbf h1:k9hqsKPh5ncKf0e3CkzvBTYXLwCYNYFb1Vtk3qnYAvk=
-github.com/apache/pulsar-client-go v0.9.1-0.20230816081803-fbee610ddcbf/go.mod h1:Ea/yiZA7plgiaWRyOuO1B0k5/hjpl1thmiKig+D9PBQ=
+github.com/apache/pulsar-client-go v0.12.0 h1:rrMlwpr6IgLRPXLRRh2vSlcw5tGV2PUSjZwmqgh2B2I=
+github.com/apache/pulsar-client-go v0.12.0/go.mod h1:dkutuH4oS2pXiGm+Ti7fQZ4MRjrMPZ8IJeEGAWMeckk=
 github.com/apparentlymart/go-cidr v1.1.0 h1:2mAhrMoF+nhXqxTzSZMUzDHkLjmIHC+Zzn4tdgBZjnU=
 github.com/apparentlymart/go-cidr v1.1.0/go.mod h1:EBcsNrHc3zQeuaeCeCtQruQm+n9/YjEn/vI25Lg7Gwc=
 github.com/apparentlymart/go-dump v0.0.0-20180507223929-23540a00eaa3/go.mod h1:oL81AME2rN47vu18xqj1S1jPIPuN7afo62yKTNn3XMM=

pulsar/resource_pulsar_namespace.go

@@ -230,6 +230,29 @@ func resourcePulsarNamespace() *schema.Resource {
 					},
 				},
 			},
+			"topic_auto_creation": {
+				Type:     schema.TypeSet,
+				Optional: true,
+				MaxItems: 1,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"enable": {
+							Type:     schema.TypeBool,
+							Required: true,
+						},
+						"type": {
+							Type:         schema.TypeString,
+							Required:     false,
+							ValidateFunc: validatePartitionedTopicType,
+						},
+						"partitions": {
+							Type:     schema.TypeInt,
+							Required: false,
+						},
+					},
+				},
+				Set: topicAutoCreationPoliciesToHash,
+			},
 		},
 	}
 }
@@ -411,6 +434,21 @@ func resourcePulsarNamespaceRead(ctx context.Context, d *schema.ResourceData, me
 		setPermissionGrant(d, grants)
 	}
 
+	if topicAutoCreation, ok := d.GetOk("topic_auto_creation"); ok && topicAutoCreation.(*schema.Set).Len() > 0 {
+		autoCreation, err := client.GetTopicAutoCreation(*ns)
+		if err != nil {
+			return diag.FromErr(fmt.Errorf("ERROR_READ_NAMESPACE: GetTopicAutoCreation: %w", err))
+		}
+
+		_ = d.Set("topic_auto_creation", schema.NewSet(topicAutoCreationPoliciesToHash, []interface{}{
+			map[string]interface{}{
+				"enable":     autoCreation.Allow,
+				"type":       autoCreation.Type,
+				"partitions": autoCreation.Partitions,
+			},
+		}))
+	}
+
 	return nil
 }
 
@@ -426,6 +464,7 @@ func resourcePulsarNamespaceUpdate(ctx context.Context, d *schema.ResourceData,
 	dispatchRateConfig := d.Get("dispatch_rate").(*schema.Set)
 	persistencePoliciesConfig := d.Get("persistence_policies").(*schema.Set)
 	permissionGrantConfig := d.Get("permission_grant").(*schema.Set)
+	topicAutoCreation := d.Get("topic_auto_creation").(*schema.Set)
 
 	nsName, err := utils.GetNameSpaceName(tenant, namespace)
 	if err != nil {
@@ -562,6 +601,17 @@ func resourcePulsarNamespaceUpdate(ctx context.Context, d *schema.ResourceData,
 		}
 	}
 
+	if topicAutoCreation.Len() > 0 {
+		topicAutoCreationPolicy := unmarshalTopicAutoCreation(topicAutoCreation)
+		if err = client.SetTopicAutoCreation(*nsName, *topicAutoCreationPolicy); err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("SetTopicAutoCreation: %w", err))
+		}
+	} else { // remove the topicAutoCreation
+		if err = client.RemoveTopicAutoCreation(*nsName); err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("RemoveTopicAutoCreation: %w", err))
+		}
+	}
+
 	if errs != nil {
 		return diag.FromErr(fmt.Errorf("ERROR_UPDATE_NAMESPACE_CONFIG: %w", errs))
 	}
@@ -591,6 +641,7 @@ func resourcePulsarNamespaceDelete(ctx context.Context, d *schema.ResourceData,
 	_ = d.Set("dispatch_rate", nil)
 	_ = d.Set("persistence_policies", nil)
 	_ = d.Set("permission_grant", nil)
+	_ = d.Set("topic_auto_creation", nil)
 
 	return nil
 }
@@ -644,6 +695,17 @@ func persistencePoliciesToHash(v interface{}) int {
 	return hashcode.String(buf.String())
 }
 
+func topicAutoCreationPoliciesToHash(v interface{}) int {
+	var buf bytes.Buffer
+	m := v.(map[string]interface{})
+
+	buf.WriteString(fmt.Sprintf("%t-", m["enable"].(bool)))
+	buf.WriteString(fmt.Sprintf("%s-", m["type"].(string)))
+	buf.WriteString(fmt.Sprintf("%d-", m["partitions"].(int)))
+
+	return hashcode.String(buf.String())
+}
+
 func unmarshalDispatchRate(v *schema.Set) *utils.DispatchRate {
 	var dispatchRate utils.DispatchRate
 
@@ -710,3 +772,17 @@ func unmarshalPersistencePolicies(v *schema.Set) *utils.PersistencePolicies {
 
 	return &persPolicies
 }
+
+func unmarshalTopicAutoCreation(v *schema.Set) *utils.TopicAutoCreationConfig {
+	var topicAutoCreation utils.TopicAutoCreationConfig
+
+	for _, policy := range v.List() {
+		data := policy.(map[string]interface{})
+
+		topicAutoCreation.Allow = data["allow"].(bool)
+		topicAutoCreation.Type = utils.TopicType(data["type"].(string))
+		topicAutoCreation.Partitions = data["partitions"].(*int)
+	}
+
+	return &topicAutoCreation
+}

pulsar/validate_helpers.go

@@ -50,3 +50,12 @@ func validateAuthAction(val interface{}, key string) (warns []string, errs []err
 	}
 	return
 }
+
+func validatePartitionedTopicType(val interface{}, key string) (warns []string, errs []error) {
+	v := val.(string)
+	_, err := utils.ParseTopicType(v)
+	if err != nil {
+		errs = append(errs, fmt.Errorf("%q must be a valid topic type (got: %s): %w", key, v, err))
+	}
+	return
+}