make pulsar topic controller as stateful reconciler (#349)

* make pulsar topic controller as statefui reconciler

* add debug logging

* address other policies

* fix header

* fix deploy

* add docs and warn about compaction subscription

* fix state annotation

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

docs/pulsar_topic.md

@@ -29,7 +29,7 @@ The `PulsarTopic` resource defines a topic in a Pulsar cluster. It allows you to
 | `geoReplicationRefs`                | List of references to PulsarGeoReplication resources, used to enable geo-replication at the topic level.                                                                                | No       |
 | `replicationClusters`               | List of clusters to which the topic is replicated. Use only if replicating clusters within the same Pulsar instance.                                                                    | No       |
 | `deduplication`                     | whether to enable message deduplication for the topic.                                                                                                                                  | No       |
-| `compactionThreshold`               | Size threshold in bytes for automatic topic compaction. When the topic reaches this size, compaction will be triggered automatically.                                                   | No       |
+| `compactionThreshold`               | Size threshold in bytes for automatic topic compaction. When the topic reaches this size, compaction will be triggered automatically. If you later disable compaction (removing this field or setting it to `0`), manually delete the `__compaction` subscription to avoid backlog accumulation. | No       |
 | `persistencePolicies`               | Persistence configuration for the topic, controlling how data is stored and replicated in BookKeeper. See [persistencePolicies](#persistencePolicies) for more details.               | No       |
 | `delayedDelivery`                   | Delayed delivery policy for the topic, allowing messages to be delivered with a delay. See [delayedDelivery](#delayedDelivery) for more details.                                       | No       |
 | `dispatchRate`                      | Message dispatch rate limiting policy for the topic, controlling the rate at which messages are delivered to consumers. See [dispatchRate](#dispatchRate) for more details.            | No       |
@@ -548,4 +548,4 @@ This example demonstrates a production-ready topic configuration with:
 - Message size limits (1MB maximum)
 - Consumer and subscription limits per topic
 - Custom properties for metadata tracking
-- Controlled auto-subscription creation (disabled for production)
+- Controlled auto-subscription creation (disabled for production)

pkg/admin/dummy.go

@@ -96,6 +96,116 @@ func (d *DummyPulsarAdmin) SetTopicClusters(string, *bool, []string) error {
 	return nil
 }
 
+// SetTopicCompactionThreshold is a fake implementation of SetTopicCompactionThreshold
+func (d *DummyPulsarAdmin) SetTopicCompactionThreshold(string, *bool, int64) error {
+	return nil
+}
+
+// RemoveTopicCompactionThreshold is a fake implementation of RemoveTopicCompactionThreshold
+func (d *DummyPulsarAdmin) RemoveTopicCompactionThreshold(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMessageTTL(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxProducers(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxConsumers(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxUnackedMessagesPerConsumer(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxUnackedMessagesPerSubscription(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicRetention(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicBacklogQuota(string, *bool, string) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicDeduplicationStatus(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicPersistence(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicDelayedDelivery(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicDispatchRate(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicPublishRate(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicInactiveTopicPolicies(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicSubscribeRate(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxMessageSize(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxConsumersPerSubscription(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicMaxSubscriptionsPerTopic(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicSchemaValidationEnforced(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicSubscriptionDispatchRate(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicReplicatorDispatchRate(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicDeduplicationSnapshotInterval(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicOffloadPolicies(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicAutoSubscriptionCreation(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicSchemaCompatibilityStrategy(string, *bool) error {
+	return nil
+}
+
+func (d *DummyPulsarAdmin) RemoveTopicProperty(string, *bool, string) error {
+	return nil
+}
+
 // Close is a fake implements of Close
 func (d *DummyPulsarAdmin) Close() error {
 	return nil