support topic compaction (#331)

* support topic compaction

* fix test

* more error log

* fix topic name

* enable topicLevelPoliciesEnabled

* Add logging for topic application errors and update test timeout settings

- Added logging for creation and policy errors during topic application in `reconcile_topic.go`.
- Increased timeout settings in tests for resource updates to ensure stability and reliability.

* fix test

Author: freeznet

.ci/clusters/values.yaml

@@ -121,6 +121,7 @@ broker:
     managedLedgerDefaultAckQuorum: "1"
     enablePackagesManagement: "true"
     PULSAR_PREFIX_enablePackagesManagement: "true"
+    PULSAR_PREFIX_topicLevelPoliciesEnabled: "true"
   resources:
     requests:
       memory: 256Mi

api/v1alpha1/pulsartopic_types.go

@@ -131,6 +131,11 @@ type PulsarTopicSpec struct {
 	// Deduplication controls whether to enable message deduplication for the topic.
 	// +optional
 	Deduplication *bool `json:"deduplication,omitempty"`
+
+	// CompactionThreshold specifies the size threshold in bytes for automatic topic compaction.
+	// When the topic reaches this size, compaction will be triggered automatically.
+	// +optional
+	CompactionThreshold *int64 `json:"compactionThreshold,omitempty"`
 }
 
 // SchemaInfo defines the Pulsar Schema for a topic.

api/v1alpha1/zz_generated.deepcopy.go

@@ -98,6 +98,12 @@ spec:
                   BacklogQuotaRetentionPolicy specifies the retention policy for messages when backlog quota is exceeded.
                   Valid values are "producer_request_hold", "producer_exception", or "consumer_backlog_eviction".
                 type: string
+              compactionThreshold:
+                description: |-
+                  CompactionThreshold specifies the size threshold in bytes for automatic topic compaction.
+                  When the topic reaches this size, compaction will be triggered automatically.
+                format: int64
+                type: integer
               connectionRef:
                 description: |-
                   ConnectionRef is the reference to the PulsarConnection resource

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

@@ -33,4 +33,5 @@ spec:
   # backlogQuotaLimitTime: 24h
   # backlogQuotaLimitSize: 1Gi
   # backlogQuotaRetentionPolicy: producer_request_hold
+  # compactionThreshold: 104857600  # 100MB in bytes
   lifecyclePolicy: CleanUpAfterDeletion

config/samples/resource_v1alpha1_pulsartopic.yaml

@@ -296,6 +296,13 @@ func (p *PulsarAdminClient) applyTopicPolicies(topicName *utils.TopicName, param
 		}
 	}
 
+	if params.CompactionThreshold != nil {
+		err = p.adminClient.Topics().SetCompactionThreshold(*topicName, *params.CompactionThreshold)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

pkg/admin/impl.go

@@ -76,6 +76,7 @@ type TopicParams struct {
 	BacklogQuotaRetentionPolicy       *string
 	ReplicationClusters               []string
 	Deduplication                     *bool
+	CompactionThreshold               *int64
 }
 
 // ClusterParams indicate the parameters for creating a cluster

pkg/admin/interface.go

@@ -231,6 +231,7 @@ func (r *PulsarTopicReconciler) ReconcileTopic(ctx context.Context, pulsarAdmin
 	}
 
 	creationErr, policyErr := pulsarAdmin.ApplyTopic(topic.Spec.Name, params)
+	log.Info("Apply topic", "creationErr", creationErr, "policyErr", policyErr)
 	if policyErr != nil {
 		policyErrs = append(policyErrs, policyErr)
 	}
@@ -288,6 +289,7 @@ func createTopicParams(topic *resourcev1alpha1.PulsarTopic) *admin.TopicParams {
 		BacklogQuotaLimitSize:             topic.Spec.BacklogQuotaLimitSize,
 		BacklogQuotaRetentionPolicy:       topic.Spec.BacklogQuotaRetentionPolicy,
 		Deduplication:                     topic.Spec.Deduplication,
+		CompactionThreshold:               topic.Spec.CompactionThreshold,
 	}
 }
 

pkg/connection/reconcile_topic.go

@@ -370,6 +370,91 @@ var _ = Describe("Resources", func() {
 
 		})
 
+		Context("PulsarTopic Compaction Threshold", Ordered, func() {
+			var (
+				compactionTopic     *v1alphav1.PulsarTopic
+				compactionTopicName string = "test-compaction-topic"
+				compactionThreshold int64  = 104857600 // 100MB in bytes
+			)
+
+			BeforeAll(func() {
+				compactionTopic = utils.MakePulsarTopicWithCompactionThreshold(
+					namespaceName,
+					compactionTopicName,
+					"persistent://public/default/compaction-test",
+					pconnName,
+					compactionThreshold,
+					lifecyclePolicy,
+				)
+			})
+
+			It("should create topic with compaction threshold successfully", func() {
+				err := k8sClient.Create(ctx, compactionTopic)
+				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
+			})
+
+			It("should be ready", func() {
+				Eventually(func() bool {
+					t := &v1alphav1.PulsarTopic{}
+					tns := types.NamespacedName{Namespace: namespaceName, Name: compactionTopicName}
+					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+					return v1alphav1.IsPulsarResourceReady(t)
+				}, "20s", "100ms").Should(BeTrue())
+			})
+
+			It("should update compaction threshold successfully", func() {
+				newThreshold := int64(209715200) // 200MB in bytes
+
+				topic := &v1alphav1.PulsarTopic{}
+				tns := types.NamespacedName{Namespace: namespaceName, Name: compactionTopicName}
+				Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed())
+
+				topic.Spec.CompactionThreshold = &newThreshold
+				err := k8sClient.Update(ctx, topic)
+				Expect(err).Should(Succeed())
+			})
+
+			It("should be ready after update", func() {
+				Eventually(func() bool {
+					t := &v1alphav1.PulsarTopic{}
+					tns := types.NamespacedName{Namespace: namespaceName, Name: compactionTopicName}
+					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+					return v1alphav1.IsPulsarResourceReady(t)
+				}, "20s", "100ms").Should(BeTrue())
+			})
+
+			It("should remove compaction threshold when set to nil", func() {
+				topic := &v1alphav1.PulsarTopic{}
+				tns := types.NamespacedName{Namespace: namespaceName, Name: compactionTopicName}
+				Expect(k8sClient.Get(ctx, tns, topic)).Should(Succeed())
+
+				topic.Spec.CompactionThreshold = nil
+				err := k8sClient.Update(ctx, topic)
+				Expect(err).Should(Succeed())
+			})
+
+			It("should be ready after removing threshold", func() {
+				Eventually(func() bool {
+					t := &v1alphav1.PulsarTopic{}
+					tns := types.NamespacedName{Namespace: namespaceName, Name: compactionTopicName}
+					Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+					return v1alphav1.IsPulsarResourceReady(t)
+				}, "20s", "100ms").Should(BeTrue())
+			})
+
+			AfterAll(func() {
+				// Clean up the compaction test topic after all tests complete
+				if compactionTopic != nil {
+					Eventually(func(g Gomega) {
+						t := &v1alphav1.PulsarTopic{}
+						tns := types.NamespacedName{Namespace: namespaceName, Name: compactionTopicName}
+						g.Expect(k8sClient.Get(ctx, tns, t)).Should(Succeed())
+						g.Expect(k8sClient.Delete(ctx, t)).Should(Succeed())
+					}).Should(Succeed())
+				}
+			})
+		})
+
 		Context("PulsarPermission operation", Label("Permissions"), func() {
 			It("should grant the pulsarpermission successfully", func() {
 				err := k8sClient.Create(ctx, ppermission)

tests/operator/resources_test.go

@@ -107,6 +107,24 @@ func MakePulsarTopic(namespace, name, topicName, connectionName string, policy v
 	}
 }
 
+// MakePulsarTopicWithCompactionThreshold will generate a object of PulsarTopic with compaction threshold
+func MakePulsarTopicWithCompactionThreshold(namespace, name, topicName, connectionName string, compactionThreshold int64, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarTopic {
+	return &v1alpha1.PulsarTopic{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+		Spec: v1alpha1.PulsarTopicSpec{
+			ConnectionRef: corev1.LocalObjectReference{
+				Name: connectionName,
+			},
+			Name:                topicName,
+			CompactionThreshold: &compactionThreshold,
+			LifecyclePolicy:     policy,
+		},
+	}
+}
+
 // MakePulsarPermission will generate a object of PulsarPermission
 func MakePulsarPermission(namespace, name, resourceName, connectionName string, resourceType v1alpha1.PulsarResourceType,
 	roles, actions []string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarPermission {