support topic compaction

Author: freeznet

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

@@ -288,6 +288,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,134 @@ 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
+			)
+
+			BeforeEach(func() {
+				compactionTopic = utils.MakePulsarTopicWithCompactionThreshold(
+					namespaceName,
+					compactionTopicName,
+					"persistent://cloud/stage/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 have compaction threshold set in Pulsar", func() {
+				Eventually(func(g Gomega) {
+					podName := fmt.Sprintf("%s-broker-0", brokerName)
+					containerName := fmt.Sprintf("%s-broker", brokerName)
+					stdout, _, err := utils.ExecInPod(k8sConfig, namespaceName, podName, containerName,
+						"./bin/pulsar-admin topics get-compaction-threshold "+compactionTopic.Spec.Name)
+					g.Expect(err).Should(Succeed())
+					g.Expect(stdout).Should(Not(BeEmpty()))
+					// The output should contain the threshold value in bytes
+					g.Expect(stdout).Should(ContainSubstring("104857600"))
+				}, "20s", "100ms").Should(Succeed())
+			})
+
+			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 have updated compaction threshold in Pulsar", func() {
+				Eventually(func(g Gomega) {
+					podName := fmt.Sprintf("%s-broker-0", brokerName)
+					containerName := fmt.Sprintf("%s-broker", brokerName)
+					stdout, _, err := utils.ExecInPod(k8sConfig, namespaceName, podName, containerName,
+						"./bin/pulsar-admin topics get-compaction-threshold "+compactionTopic.Spec.Name)
+					g.Expect(err).Should(Succeed())
+					g.Expect(stdout).Should(Not(BeEmpty()))
+					// The output should contain the new threshold value in bytes
+					g.Expect(stdout).Should(ContainSubstring("209715200"))
+				}, "20s", "100ms").Should(Succeed())
+			})
+
+			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())
+			})
+
+			It("should have no compaction threshold in Pulsar after removal", func() {
+				Eventually(func(g Gomega) {
+					podName := fmt.Sprintf("%s-broker-0", brokerName)
+					containerName := fmt.Sprintf("%s-broker", brokerName)
+					stdout, stderr, err := utils.ExecInPod(k8sConfig, namespaceName, podName, containerName,
+						"./bin/pulsar-admin topics get-compaction-threshold "+compactionTopic.Spec.Name)
+					// When no compaction threshold is set, the command might fail or return empty/default value
+					// We need to check both success and failure cases
+					if err != nil {
+						// Command failed, which might be expected when no threshold is set
+						g.Expect(stderr).Should(ContainSubstring("not found"))
+					} else {
+						// Command succeeded but should show no threshold or default value
+						g.Expect(stdout).Should(SatisfyAny(
+							BeEmpty(),
+							ContainSubstring("0"),
+							ContainSubstring("-1"),
+						))
+					}
+				}, "20s", "100ms").Should(Succeed())
+			})
+
+			AfterEach(func() {
+				// Clean up the compaction test topic
+				if compactionTopic != nil {
+					k8sClient.Delete(ctx, compactionTopic)
+				}
+			})
+		})
+
 		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 {