fix deploy

Author: freeznet

config/crd/kustomization.yaml

@@ -16,24 +16,25 @@
 # since it depends on service name and namespace that are out of this kustomize package.
 # It should be run by config/default
 resources:
-- bases/resource.streamnative.io_pulsarconnections.yaml
-- bases/resource.streamnative.io_pulsarnamespaces.yaml
-- bases/resource.streamnative.io_pulsartenants.yaml
-- bases/resource.streamnative.io_pulsartopics.yaml
-- bases/resource.streamnative.io_pulsarpermissions.yaml
-- bases/resource.streamnative.io_pulsargeoreplications.yaml
-- bases/resource.streamnative.io_pulsarfunctions.yaml
-- bases/resource.streamnative.io_pulsarpackages.yaml
-- bases/resource.streamnative.io_pulsarsinks.yaml
-- bases/resource.streamnative.io_pulsarsources.yaml
-- bases/resource.streamnative.io_pulsarnsisolationpolicies.yaml
-- bases/resource.streamnative.io_streamnativecloudconnections.yaml
-- bases/resource.streamnative.io_computeworkspaces.yaml
-- bases/resource.streamnative.io_computeflinkdeployments.yaml
-- bases/resource.streamnative.io_secrets.yaml
-- bases/resource.streamnative.io_serviceaccounts.yaml
-- bases/resource.streamnative.io_serviceaccountbindings.yaml
-- bases/resource.streamnative.io_apikeys.yaml
+  - bases/resource.streamnative.io_pulsarconnections.yaml
+  - bases/resource.streamnative.io_pulsarnamespaces.yaml
+  - bases/resource.streamnative.io_pulsartenants.yaml
+  - bases/resource.streamnative.io_pulsartopics.yaml
+  - bases/resource.streamnative.io_pulsarpermissions.yaml
+  - bases/resource.streamnative.io_pulsargeoreplications.yaml
+  - bases/resource.streamnative.io_pulsarfunctions.yaml
+  - bases/resource.streamnative.io_pulsarpackages.yaml
+  - bases/resource.streamnative.io_pulsarsinks.yaml
+  - bases/resource.streamnative.io_pulsarsources.yaml
+  - bases/resource.streamnative.io_pulsarnsisolationpolicies.yaml
+  - bases/resource.streamnative.io_streamnativecloudconnections.yaml
+  - bases/resource.streamnative.io_computeworkspaces.yaml
+  - bases/resource.streamnative.io_computeflinkdeployments.yaml
+  - bases/resource.streamnative.io_secrets.yaml
+  - bases/resource.streamnative.io_serviceaccounts.yaml
+  - bases/resource.streamnative.io_serviceaccountbindings.yaml
+  - bases/resource.streamnative.io_apikeys.yaml
+  - bases/resource.streamnative.io_rolebindings.yaml
 #+kubebuilder:scaffold:crdkustomizeresource
 
 patchesStrategicMerge:
@@ -59,4 +60,4 @@ patchesStrategicMerge:
 
 # the following config is for teaching kustomize how to do kustomization for CRDs.
 configurations:
-- kustomizeconfig.yaml
+  - kustomizeconfig.yaml

pkg/admin/impl.go

@@ -280,6 +280,7 @@ func (p *PulsarAdminClient) applyTopicPolicies(topicName *utils.TopicName, param
 		}
 	}
 
+	var retentionPolicy *utils.RetentionPolicies
 	if params.RetentionTime != nil || params.RetentionSize != nil {
 		retentionTime := -1
 		retentionSize := -1
@@ -301,18 +302,16 @@ func (p *PulsarAdminClient) applyTopicPolicies(topicName *utils.TopicName, param
 				retentionSize = int(params.RetentionSize.ScaledValue(resource.Mega))
 			}
 		}
-		retentionPolicy := utils.NewRetentionPolicies(retentionTime, retentionSize)
-		err = p.adminClient.Topics().SetRetention(*topicName, retentionPolicy)
-		if err != nil {
-			return err
-		}
+		policy := utils.NewRetentionPolicies(retentionTime, retentionSize)
+		retentionPolicy = &policy
 	}
 
+	var backlogQuotaPolicy *utils.BacklogQuota
+	var backlogQuotaType utils.BacklogQuotaType
 	if (params.BacklogQuotaLimitTime != nil || params.BacklogQuotaLimitSize != nil) &&
 		params.BacklogQuotaRetentionPolicy != nil {
 		backlogTime := int64(-1)
 		backlogSize := int64(-1)
-		var backlogQuotaType utils.BacklogQuotaType
 		if params.BacklogQuotaLimitTime != nil {
 			t, err := params.BacklogQuotaLimitTime.Parse()
 			if err != nil {
@@ -325,13 +324,24 @@ func (p *PulsarAdminClient) applyTopicPolicies(topicName *utils.TopicName, param
 			backlogSize = params.BacklogQuotaLimitSize.Value()
 			backlogQuotaType = utils.DestinationStorage
 		}
-		backlogQuotaPolicy := utils.BacklogQuota{
+		backlogQuotaPolicy = &utils.BacklogQuota{
 			LimitTime: backlogTime,
 			LimitSize: backlogSize,
 			Policy:    utils.RetentionPolicy(*params.BacklogQuotaRetentionPolicy),
 		}
-		err = p.adminClient.Topics().SetBacklogQuota(*topicName, backlogQuotaPolicy, backlogQuotaType)
-		if err != nil {
+	}
+
+	switch {
+	case retentionPolicy != nil && backlogQuotaPolicy != nil:
+		if err := p.applyRetentionAndBacklogPolicies(topicName, retentionPolicy, backlogQuotaPolicy, backlogQuotaType); err != nil {
+			return err
+		}
+	case retentionPolicy != nil:
+		if err := p.adminClient.Topics().SetRetention(*topicName, *retentionPolicy); err != nil {
+			return err
+		}
+	case backlogQuotaPolicy != nil:
+		if err := p.adminClient.Topics().SetBacklogQuota(*topicName, *backlogQuotaPolicy, backlogQuotaType); err != nil {
 			return err
 		}
 	}
@@ -577,6 +587,38 @@ func (p *PulsarAdminClient) applyTopicPolicies(topicName *utils.TopicName, param
 	return nil
 }
 
+func (p *PulsarAdminClient) applyRetentionAndBacklogPolicies(topicName *utils.TopicName, retention *utils.RetentionPolicies,
+	backlog *utils.BacklogQuota, backlogType utils.BacklogQuotaType) error {
+	if err := p.adminClient.Topics().SetRetention(*topicName, *retention); err != nil {
+		if !isRetentionBacklogOrderingError(err) {
+			return err
+		}
+
+		if err := p.adminClient.Topics().SetBacklogQuota(*topicName, *backlog, backlogType); err != nil {
+			return err
+		}
+
+		if err := p.adminClient.Topics().SetRetention(*topicName, *retention); err != nil {
+			return err
+		}
+
+		return nil
+	}
+
+	if err := p.adminClient.Topics().SetBacklogQuota(*topicName, *backlog, backlogType); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func isRetentionBacklogOrderingError(err error) bool {
+	if ErrorReason(err) != ReasonInvalidParameter {
+		return false
+	}
+	return strings.Contains(err.Error(), "Retention Quota must exceed configured backlog quota")
+}
+
 // GetTopicClusters get the assigned clusters of the topic to the local default cluster
 func (p *PulsarAdminClient) GetTopicClusters(name string, persistent *bool) ([]string, error) {
 	completeTopicName := MakeCompleteTopicName(name, persistent)