fix ci

Author: freeznet

api/v1alpha1/pulsarnamespace_types.go

@@ -133,6 +133,7 @@ type InactiveTopicPolicies struct {
 
 // PulsarNamespaceSpec defines the desired state of a Pulsar namespace.
 // It corresponds to the configuration options available in Pulsar's namespace admin API.
+// +kubebuilder:validation:XValidation:rule="!(has(self.backlogQuotaLimitSize) || has(self.backlogQuotaLimitTime) || has(self.backlogQuotaType)) || has(self.backlogQuotaRetentionPolicy)",message="backlogQuotaRetentionPolicy is required when configuring backlog quota"
 type PulsarNamespaceSpec struct {
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file

api/v1alpha1/pulsartopic_types.go

@@ -27,6 +27,7 @@ import (
 
 // PulsarTopicSpec defines the desired state of PulsarTopic.
 // It corresponds to the configuration options available in Pulsar's topic admin API.
+// +kubebuilder:validation:XValidation:rule="!(has(self.backlogQuotaLimitSize) || has(self.backlogQuotaLimitTime) || has(self.backlogQuotaType)) || has(self.backlogQuotaRetentionPolicy)",message="backlogQuotaRetentionPolicy is required when configuring backlog quota"
 type PulsarTopicSpec struct {
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file

charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsarnamespaces.yaml

@@ -519,6 +519,11 @@ spec:
             - connectionRef
             - name
             type: object
+            x-kubernetes-validations:
+            - message: backlogQuotaRetentionPolicy is required when configuring backlog
+                quota
+              rule: '!(has(self.backlogQuotaLimitSize) || has(self.backlogQuotaLimitTime)
+                || has(self.backlogQuotaType)) || has(self.backlogQuotaRetentionPolicy)'
           status:
             description: PulsarNamespaceStatus defines the observed state of PulsarNamespace
             properties:

charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsartopics.yaml

@@ -552,6 +552,11 @@ spec:
             - connectionRef
             - name
             type: object
+            x-kubernetes-validations:
+            - message: backlogQuotaRetentionPolicy is required when configuring backlog
+                quota
+              rule: '!(has(self.backlogQuotaLimitSize) || has(self.backlogQuotaLimitTime)
+                || has(self.backlogQuotaType)) || has(self.backlogQuotaRetentionPolicy)'
           status:
             description: PulsarTopicStatus defines the observed state of PulsarTopic
             properties:

config/crd/bases/resource.streamnative.io_pulsarnamespaces.yaml

@@ -519,6 +519,11 @@ spec:
             - connectionRef
             - name
             type: object
+            x-kubernetes-validations:
+            - message: backlogQuotaRetentionPolicy is required when configuring backlog
+                quota
+              rule: '!(has(self.backlogQuotaLimitSize) || has(self.backlogQuotaLimitTime)
+                || has(self.backlogQuotaType)) || has(self.backlogQuotaRetentionPolicy)'
           status:
             description: PulsarNamespaceStatus defines the observed state of PulsarNamespace
             properties:

config/crd/bases/resource.streamnative.io_pulsartopics.yaml

@@ -552,6 +552,11 @@ spec:
             - connectionRef
             - name
             type: object
+            x-kubernetes-validations:
+            - message: backlogQuotaRetentionPolicy is required when configuring backlog
+                quota
+              rule: '!(has(self.backlogQuotaLimitSize) || has(self.backlogQuotaLimitTime)
+                || has(self.backlogQuotaType)) || has(self.backlogQuotaRetentionPolicy)'
           status:
             description: PulsarTopicStatus defines the observed state of PulsarTopic
             properties:

pkg/admin/impl.go

@@ -613,18 +613,28 @@ func buildBacklogQuota(limitTime *rutils.Duration, limitSize *resource.Quantity,
 	if retentionPolicyStr == nil {
 		return nil, "", fmt.Errorf("backlogQuotaRetentionPolicy is required when configuring backlog quota")
 	}
+	if limitTime == nil && limitSize == nil {
+		return nil, "", fmt.Errorf("backlogQuotaLimitTime or backlogQuotaLimitSize is required when configuring backlog quota")
+	}
 	retentionPolicy, err := utils.ParseRetentionPolicy(*retentionPolicyStr)
 	if err != nil {
 		return nil, "", err
 	}
 
-	backlogQuotaType := utils.DestinationStorage
-	if backlogQuotaTypeStr != nil {
+	var backlogQuotaType utils.BacklogQuotaType
+	switch {
+	case backlogQuotaTypeStr != nil:
 		parsedType, err := utils.ParseBacklogQuotaType(*backlogQuotaTypeStr)
 		if err != nil {
 			return nil, "", err
 		}
 		backlogQuotaType = parsedType
+	case limitSize != nil:
+		backlogQuotaType = utils.DestinationStorage
+	case limitTime != nil:
+		backlogQuotaType = utils.MessageAge
+	default:
+		backlogQuotaType = utils.DestinationStorage
 	}
 
 	backlogQuota := utils.BacklogQuota{

tests/utils/spec.go

@@ -67,6 +67,8 @@ func MakePulsarNamespace(namespace, name, namespaceName, connectionName string,
 	var du rsutils.Duration = "1d"
 	limitTime := &du
 	ttl := &du
+	backlogPolicy := ptr.To("producer_request_hold")
+	backlogType := ptr.To("destination_storage")
 	return &v1alpha1.PulsarNamespace{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: namespace,
@@ -77,10 +79,12 @@ func MakePulsarNamespace(namespace, name, namespaceName, connectionName string,
 			ConnectionRef: corev1.LocalObjectReference{
 				Name: connectionName,
 			},
-			BacklogQuotaLimitTime: limitTime,
-			BacklogQuotaLimitSize: &backlogSize,
-			Bundles:               &bundle,
-			MessageTTL:            ttl,
+			BacklogQuotaLimitTime:       limitTime,
+			BacklogQuotaLimitSize:       &backlogSize,
+			BacklogQuotaRetentionPolicy: backlogPolicy,
+			BacklogQuotaType:            backlogType,
+			Bundles:                     &bundle,
+			MessageTTL:                  ttl,
 			TopicAutoCreationConfig: &v1alpha1.TopicAutoCreationConfig{
 				Allow:      true,
 				Type:       "partitioned",
@@ -97,6 +101,8 @@ func MakePulsarNamespaceWithRateLimiting(namespace, name, namespaceName, connect
 	var du rsutils.Duration = "12h"
 	limitTime := &du
 	ttl := &du
+	backlogPolicy := ptr.To("producer_request_hold")
+	backlogType := ptr.To("destination_storage")
 
 	return &v1alpha1.PulsarNamespace{
 		ObjectMeta: metav1.ObjectMeta{
@@ -108,11 +114,13 @@ func MakePulsarNamespaceWithRateLimiting(namespace, name, namespaceName, connect
 			ConnectionRef: corev1.LocalObjectReference{
 				Name: connectionName,
 			},
-			BacklogQuotaLimitTime: limitTime,
-			BacklogQuotaLimitSize: &backlogSize,
-			Bundles:               &bundle,
-			MessageTTL:            ttl,
-			LifecyclePolicy:       policy,
+			BacklogQuotaLimitTime:       limitTime,
+			BacklogQuotaLimitSize:       &backlogSize,
+			BacklogQuotaRetentionPolicy: backlogPolicy,
+			BacklogQuotaType:            backlogType,
+			Bundles:                     &bundle,
+			MessageTTL:                  ttl,
+			LifecyclePolicy:             policy,
 
 			// Rate limiting configurations
 			DispatchRate: &v1alpha1.DispatchRate{