feat: enhance Topic with missing policies (#337)

* feat: add new topic policies including persistence, delivery, dispatch, publish, and inactive topic management

* feat: add comprehensive tests for PulsarTopic policies including persistence, delayed delivery, dispatch rate, publish rate, and inactive topic policies

* feat: add advanced topic-level policies to PulsarTopic documentation, including compaction, persistence, delayed delivery, dispatch rate, publish rate, and inactive topic management

Author: freeznet

api/v1alpha1/pulsartopic_types.go

@@ -136,6 +136,42 @@ type PulsarTopicSpec struct {
 	// When the topic reaches this size, compaction will be triggered automatically.
 	// +optional
 	CompactionThreshold *int64 `json:"compactionThreshold,omitempty"`
+
+	// PersistencePolicies defines the persistence configuration for the topic.
+	// This controls how data is stored and replicated in BookKeeper.
+	// +optional
+	PersistencePolicies *PersistencePolicies `json:"persistencePolicies,omitempty"`
+
+	// DelayedDelivery defines the delayed delivery policy for the topic.
+	// This allows messages to be delivered with a delay.
+	// +optional
+	DelayedDelivery *DelayedDeliveryData `json:"delayedDelivery,omitempty"`
+
+	// DispatchRate defines the message dispatch rate limiting policy for the topic.
+	// This controls the rate at which messages are delivered to consumers.
+	// +optional
+	DispatchRate *DispatchRate `json:"dispatchRate,omitempty"`
+
+	// PublishRate defines the message publish rate limiting policy for the topic.
+	// This controls the rate at which producers can publish messages.
+	// +optional
+	PublishRate *PublishRate `json:"publishRate,omitempty"`
+
+	// InactiveTopicPolicies defines the inactive topic cleanup policy for the topic.
+	// This controls how inactive topics are automatically cleaned up.
+	// +optional
+	InactiveTopicPolicies *InactiveTopicPolicies `json:"inactiveTopicPolicies,omitempty"`
+}
+
+// DelayedDeliveryData defines the delayed delivery policy for a topic
+type DelayedDeliveryData struct {
+	// Active determines whether delayed delivery is enabled for the topic
+	// +optional
+	Active *bool `json:"active,omitempty"`
+
+	// TickTimeMillis specifies the tick time for delayed message delivery in milliseconds
+	// +optional
+	TickTimeMillis *int64 `json:"tickTimeMillis,omitempty"`
 }
 
 // SchemaInfo defines the Pulsar Schema for a topic.

api/v1alpha1/zz_generated.deepcopy.go

@@ -126,6 +126,46 @@ spec:
                 description: Deduplication controls whether to enable message deduplication
                   for the topic.
                 type: boolean
+              delayedDelivery:
+                description: |-
+                  DelayedDelivery defines the delayed delivery policy for the topic.
+                  This allows messages to be delivered with a delay.
+                properties:
+                  active:
+                    description: Active determines whether delayed delivery is enabled
+                      for the topic
+                    type: boolean
+                  tickTimeMillis:
+                    description: TickTimeMillis specifies the tick time for delayed
+                      message delivery in milliseconds
+                    format: int64
+                    type: integer
+                type: object
+              dispatchRate:
+                description: |-
+                  DispatchRate defines the message dispatch rate limiting policy for the topic.
+                  This controls the rate at which messages are delivered to consumers.
+                properties:
+                  dispatchThrottlingRateInByte:
+                    description: |-
+                      DispatchThrottlingRateInByte specifies the maximum number of bytes per second allowed
+                      -1 means unlimited
+                    format: int64
+                    type: integer
+                  dispatchThrottlingRateInMsg:
+                    description: |-
+                      DispatchThrottlingRateInMsg specifies the maximum number of messages per second allowed
+                      -1 means unlimited
+                    format: int32
+                    type: integer
+                  ratePeriodInSecond:
+                    default: 1
+                    description: RatePeriodInSecond specifies the time window in seconds
+                      for rate calculation
+                    format: int32
+                    minimum: 1
+                    type: integer
+                type: object
               geoReplicationRefs:
                 description: |-
                   GeoReplicationRefs is a list of references to PulsarGeoReplication resources,
@@ -151,6 +191,30 @@ spec:
                   type: object
                   x-kubernetes-map-type: atomic
                 type: array
+              inactiveTopicPolicies:
+                description: |-
+                  InactiveTopicPolicies defines the inactive topic cleanup policy for the topic.
+                  This controls how inactive topics are automatically cleaned up.
+                properties:
+                  deleteWhileInactive:
+                    description: DeleteWhileInactive specifies whether to delete topics
+                      while they are inactive
+                    type: boolean
+                  inactiveTopicDeleteMode:
+                    description: |-
+                      InactiveTopicDeleteMode specifies how inactive topics should be handled
+                      Valid values: "delete_when_no_subscriptions", "delete_when_subscriptions_caught_up"
+                    enum:
+                    - delete_when_no_subscriptions
+                    - delete_when_subscriptions_caught_up
+                    type: string
+                  maxInactiveDurationInSeconds:
+                    description: MaxInactiveDurationInSeconds specifies how long a
+                      topic can remain inactive before being deleted
+                    format: int32
+                    minimum: 1
+                    type: integer
+                type: object
               lifecyclePolicy:
                 description: |-
                   LifecyclePolicy determines whether to keep or delete the Pulsar topic
@@ -196,12 +260,63 @@ spec:
                   Set to 0 for a non-partitioned topic.
                 format: int32
                 type: integer
+              persistencePolicies:
+                description: |-
+                  PersistencePolicies defines the persistence configuration for the topic.
+                  This controls how data is stored and replicated in BookKeeper.
+                properties:
+                  bookkeeperAckQuorum:
+                    description: |-
+                      BookkeeperAckQuorum specifies the number of replicas to wait for acknowledgment
+                      Must be <= BookkeeperWriteQuorum
+                    format: int32
+                    minimum: 1
+                    type: integer
+                  bookkeeperEnsemble:
+                    description: |-
+                      BookkeeperEnsemble specifies the number of bookies to use for a ledger
+                      This determines the replication factor for storing data
+                    format: int32
+                    minimum: 1
+                    type: integer
+                  bookkeeperWriteQuorum:
+                    description: |-
+                      BookkeeperWriteQuorum specifies the number of replicas to write for each entry
+                      Must be <= BookkeeperEnsemble
+                    format: int32
+                    minimum: 1
+                    type: integer
+                  managedLedgerMaxMarkDeleteRate:
+                    description: |-
+                      ManagedLedgerMaxMarkDeleteRate specifies the maximum rate at which mark-delete operations can be performed
+                      This helps control the rate of acknowledgment processing
+                      Value should be a decimal number as string (e.g., "1.5", "2.0")
+                    type: string
+                type: object
               persistent:
                 default: true
                 description: |-
                   Persistent determines if the topic is persistent (true) or non-persistent (false).
                   Defaults to true if not specified.
                 type: boolean
+              publishRate:
+                description: |-
+                  PublishRate defines the message publish rate limiting policy for the topic.
+                  This controls the rate at which producers can publish messages.
+                properties:
+                  publishThrottlingRateInByte:
+                    description: |-
+                      PublishThrottlingRateInByte specifies the maximum number of bytes per second that producers can publish
+                      -1 means unlimited
+                    format: int64
+                    type: integer
+                  publishThrottlingRateInMsg:
+                    description: |-
+                      PublishThrottlingRateInMsg specifies the maximum number of messages per second that producers can publish
+                      -1 means unlimited
+                    format: int32
+                    type: integer
+                type: object
               replicationClusters:
                 description: |-
                   ReplicationClusters is the list of clusters to which the topic is replicated