feat: enhance Topic with missing policies #337

freeznet · 2025-07-25T01:44:12Z

(If this PR fixes a github issue, please add Fixes #<xyz>.)

Fixes #

(or if this PR is one task of a github issue, please add Master Issue: #<xyz> to link to the master issue.)

Master Issue: #

Motivation

Explain here the context, and why you're making that change. What is the problem you're trying to solve.

Modifications

Describe the modifications you've done.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

Documentation

Check the box below.

Need to update docs?

doc-required

(If you need help on updating docs, create a doc issue)
no-need-doc

(Please explain why)
doc

(If this PR contains doc changes)

…h, publish, and inactive topic management

…istence, delayed delivery, dispatch rate, publish rate, and inactive topic policies

… including compaction, persistence, delayed delivery, dispatch rate, publish rate, and inactive topic management

github-actions · 2025-07-25T01:44:20Z

@freeznet:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

Copilot

Pull Request Overview

This PR enhances the PulsarTopic resource by adding support for several missing topic-level policies. The changes enable users to configure advanced persistence settings, delayed delivery, rate limiting, and inactive topic management policies directly through the Kubernetes resource definition.

Key changes include:

Added new policy structs for PersistencePolicies, DelayedDeliveryData, DispatchRate, PublishRate, and InactiveTopicPolicies
Extended PulsarTopic spec to include these new policy fields
Implemented backend support in admin client and reconciliation logic
Added comprehensive test coverage for all new policies
Updated documentation with detailed usage examples

Reviewed Changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 12 comments.

Show a summary per file

File	Description
`api/v1alpha1/pulsartopic_types.go`	Added new policy type definitions and extended PulsarTopicSpec
`config/crd/bases/resource.streamnative.io_pulsartopics.yaml`	Updated CRD schema with new policy field definitions
`pkg/admin/interface.go`	Extended TopicParams struct with new policy fields
`pkg/admin/impl.go`	Implemented policy application logic in admin client
`pkg/connection/reconcile_topic.go`	Added new policy fields to topic parameter creation
`tests/utils/spec.go`	Added utility functions for creating topics with different policy configurations
`tests/operator/resources_test.go`	Added comprehensive integration tests for all new policies
`docs/pulsar_topic.md`	Updated documentation with detailed policy descriptions and examples
`api/v1alpha1/zz_generated.deepcopy.go`	Auto-generated deepcopy methods for new types
`go.work.sum`	Updated go module checksums

Copilot · 2025-07-25T02:31:05Z

pkg/admin/impl.go

+	}
+
+	// Handle inactive topic policies
+	if params.InactiveTopicPolicies != nil {


Potential nil pointer dereference. The InactiveTopicDeleteMode field is a pointer and should be checked for nil before dereferencing.

Suggested change

if params.InactiveTopicPolicies != nil {

if params.InactiveTopicPolicies != nil {

if params.InactiveTopicPolicies.InactiveTopicDeleteMode == nil {

return fmt.Errorf("InactiveTopicDeleteMode is nil in InactiveTopicPolicies")

}

Copilot · 2025-07-25T02:31:05Z

pkg/admin/impl.go

+		inactiveTopicPolicies := utils.InactiveTopicPolicies{
+			InactiveTopicDeleteMode:    &deleteMode,
+			MaxInactiveDurationSeconds: int(*params.InactiveTopicPolicies.MaxInactiveDurationInSeconds),


Potential nil pointer dereference. The MaxInactiveDurationInSeconds field is a pointer and should be checked for nil before dereferencing.

Suggested change

inactiveTopicPolicies := utils.InactiveTopicPolicies{

InactiveTopicDeleteMode: &deleteMode,

MaxInactiveDurationSeconds: int(*params.InactiveTopicPolicies.MaxInactiveDurationInSeconds),

var maxInactiveDurationSeconds int

if params.InactiveTopicPolicies.MaxInactiveDurationInSeconds != nil {

maxInactiveDurationSeconds = int(*params.InactiveTopicPolicies.MaxInactiveDurationInSeconds)

} else {

maxInactiveDurationSeconds = 0 // Default value if nil

}

inactiveTopicPolicies := utils.InactiveTopicPolicies{

InactiveTopicDeleteMode: &deleteMode,

MaxInactiveDurationSeconds: maxInactiveDurationSeconds,

Copilot · 2025-07-25T02:31:05Z

pkg/admin/impl.go

+			DeleteWhileInactive:        *params.InactiveTopicPolicies.DeleteWhileInactive,
+		}


Potential nil pointer dereference. The DeleteWhileInactive field is a pointer and should be checked for nil before dereferencing.

Suggested change

DeleteWhileInactive: *params.InactiveTopicPolicies.DeleteWhileInactive,

}

DeleteWhileInactive: false, // Default value

}

if params.InactiveTopicPolicies.DeleteWhileInactive != nil {

inactiveTopicPolicies.DeleteWhileInactive = *params.InactiveTopicPolicies.DeleteWhileInactive

}

Copilot · 2025-07-25T02:31:06Z

pkg/admin/impl.go

+				return err
+			}
+		}
+


Potential nil pointer dereference. The BookkeeperEnsemble field is a pointer and should be checked for nil before dereferencing.

Suggested change

// Check for nil pointers in PersistencePolicies fields

if params.PersistencePolicies.BookkeeperEnsemble == nil {

return errors.New("BookkeeperEnsemble cannot be nil")

}

Copilot · 2025-07-25T02:31:06Z

pkg/admin/impl.go

+
+		persistenceData := utils.PersistenceData{
+			BookkeeperEnsemble:             int64(*params.PersistencePolicies.BookkeeperEnsemble),
+			BookkeeperWriteQuorum:          int64(*params.PersistencePolicies.BookkeeperWriteQuorum),


Potential nil pointer dereference. The BookkeeperWriteQuorum field is a pointer and should be checked for nil before dereferencing.

Copilot · 2025-07-25T02:31:06Z

pkg/admin/impl.go

+		dispatchRateData := utils.DispatchRateData{
+			DispatchThrottlingRateInMsg:  int64(*params.DispatchRate.DispatchThrottlingRateInMsg),


Potential nil pointer dereference. The DispatchThrottlingRateInMsg field is a pointer and should be checked for nil before dereferencing.

Suggested change

dispatchRateData := utils.DispatchRateData{

DispatchThrottlingRateInMsg: int64(*params.DispatchRate.DispatchThrottlingRateInMsg),

var dispatchThrottlingRateInMsg int64

if params.DispatchRate.DispatchThrottlingRateInMsg != nil {

dispatchThrottlingRateInMsg = int64(*params.DispatchRate.DispatchThrottlingRateInMsg)

}

dispatchRateData := utils.DispatchRateData{

DispatchThrottlingRateInMsg: dispatchThrottlingRateInMsg,

Copilot · 2025-07-25T02:31:07Z

pkg/admin/impl.go

+			DispatchThrottlingRateInMsg:  int64(*params.DispatchRate.DispatchThrottlingRateInMsg),
+			DispatchThrottlingRateInByte: *params.DispatchRate.DispatchThrottlingRateInByte,
+			RatePeriodInSecond:           int64(*params.DispatchRate.RatePeriodInSecond),
+		}


Potential nil pointer dereference. The DispatchThrottlingRateInByte field is a pointer and should be checked for nil before dereferencing.

Suggested change

DispatchThrottlingRateInMsg: int64(*params.DispatchRate.DispatchThrottlingRateInMsg),

DispatchThrottlingRateInByte: *params.DispatchRate.DispatchThrottlingRateInByte,

RatePeriodInSecond: int64(*params.DispatchRate.RatePeriodInSecond),

}

DispatchThrottlingRateInMsg: int64(*params.DispatchRate.DispatchThrottlingRateInMsg),

}

if params.DispatchRate.DispatchThrottlingRateInByte != nil {

dispatchRateData.DispatchThrottlingRateInByte = *params.DispatchRate.DispatchThrottlingRateInByte

}

dispatchRateData.RatePeriodInSecond = int64(*params.DispatchRate.RatePeriodInSecond)

Copilot · 2025-07-25T02:31:07Z

pkg/admin/impl.go

+		dispatchRateData := utils.DispatchRateData{
+			DispatchThrottlingRateInMsg:  int64(*params.DispatchRate.DispatchThrottlingRateInMsg),
+			DispatchThrottlingRateInByte: *params.DispatchRate.DispatchThrottlingRateInByte,
+			RatePeriodInSecond:           int64(*params.DispatchRate.RatePeriodInSecond),


Potential nil pointer dereference. The RatePeriodInSecond field is a pointer and should be checked for nil before dereferencing.

Suggested change

RatePeriodInSecond: int64(*params.DispatchRate.RatePeriodInSecond),

RatePeriodInSecond: func() int64 {

if params.DispatchRate.RatePeriodInSecond != nil {

return int64(*params.DispatchRate.RatePeriodInSecond)

}

return 0 // Default value if RatePeriodInSecond is nil

}(),

Copilot · 2025-07-25T02:31:07Z

pkg/admin/impl.go

+	// Handle publish rate
+	if params.PublishRate != nil {
+		publishRateData := utils.PublishRateData{
+			PublishThrottlingRateInMsg:  int64(*params.PublishRate.PublishThrottlingRateInMsg),


Potential nil pointer dereference. The PublishThrottlingRateInMsg field is a pointer and should be checked for nil before dereferencing.

Suggested change

PublishThrottlingRateInMsg: int64(*params.PublishRate.PublishThrottlingRateInMsg),

PublishThrottlingRateInMsg: func() int64 {

if params.PublishRate.PublishThrottlingRateInMsg != nil {

return int64(*params.PublishRate.PublishThrottlingRateInMsg)

}

return 0

}(),

Copilot · 2025-07-25T02:31:07Z

pkg/admin/impl.go

+			PublishThrottlingRateInMsg:  int64(*params.PublishRate.PublishThrottlingRateInMsg),
+			PublishThrottlingRateInByte: *params.PublishRate.PublishThrottlingRateInByte,


Potential nil pointer dereference. The PublishThrottlingRateInByte field is a pointer and should be checked for nil before dereferencing.

Suggested change

PublishThrottlingRateInMsg: int64(*params.PublishRate.PublishThrottlingRateInMsg),

PublishThrottlingRateInByte: *params.PublishRate.PublishThrottlingRateInByte,

PublishThrottlingRateInMsg: int64(*params.PublishRate.PublishThrottlingRateInMsg),

PublishThrottlingRateInByte: func() int64 {

if params.PublishRate.PublishThrottlingRateInByte != nil {

return *params.PublishRate.PublishThrottlingRateInByte

}

return 0

}(),

freeznet added 3 commits July 25, 2025 09:20

feat: add new topic policies including persistence, delivery, dispatc…

4cf4a4b

…h, publish, and inactive topic management

feat: add comprehensive tests for PulsarTopic policies including pers…

4913932

…istence, delayed delivery, dispatch rate, publish rate, and inactive topic policies

feat: add advanced topic-level policies to PulsarTopic documentation,…

92f5c67

… including compaction, persistence, delayed delivery, dispatch rate, publish rate, and inactive topic management

freeznet self-assigned this Jul 25, 2025

freeznet requested review from a team as code owners July 25, 2025 01:44

github-actions bot added the doc-info-missing This pr needs to mark a document option in description label Jul 25, 2025

freeznet mentioned this pull request Jul 25, 2025

feat[namespace+topic]: support schema related policies #72

Closed

labuladong requested a review from Copilot July 25, 2025 02:28

Copilot AI reviewed Jul 25, 2025

View reviewed changes

labuladong approved these changes Jul 25, 2025

View reviewed changes

maxsxu approved these changes Jul 25, 2025

View reviewed changes

freeznet merged commit 8d8c6f9 into main Jul 28, 2025
6 checks passed

freeznet deleted the freeznet/align-topic-level-polcies branch July 28, 2025 02:33

-		inactiveTopicPolicies := utils.InactiveTopicPolicies{
-			InactiveTopicDeleteMode:    &deleteMode,
-			MaxInactiveDurationSeconds: int(*params.InactiveTopicPolicies.MaxInactiveDurationInSeconds),
+		var maxInactiveDurationSeconds int
+		if params.InactiveTopicPolicies.MaxInactiveDurationInSeconds != nil {
+			maxInactiveDurationSeconds = int(*params.InactiveTopicPolicies.MaxInactiveDurationInSeconds)
+		} else {
+			maxInactiveDurationSeconds = 0 // Default value if nil
+		}
+		inactiveTopicPolicies := utils.InactiveTopicPolicies{
+			InactiveTopicDeleteMode:    &deleteMode,
+			MaxInactiveDurationSeconds: maxInactiveDurationSeconds,

		DeleteWhileInactive: *params.InactiveTopicPolicies.DeleteWhileInactive,
		}

-			DeleteWhileInactive:        *params.InactiveTopicPolicies.DeleteWhileInactive,
-		}
+			DeleteWhileInactive:        false, // Default value
+		}
+		if params.InactiveTopicPolicies.DeleteWhileInactive != nil {
+			inactiveTopicPolicies.DeleteWhileInactive = *params.InactiveTopicPolicies.DeleteWhileInactive
+		}

+		// Check for nil pointers in PersistencePolicies fields
+		if params.PersistencePolicies.BookkeeperEnsemble == nil {
+			return errors.New("BookkeeperEnsemble cannot be nil")
+		}

		dispatchRateData := utils.DispatchRateData{
		DispatchThrottlingRateInMsg: int64(*params.DispatchRate.DispatchThrottlingRateInMsg),

-		dispatchRateData := utils.DispatchRateData{
-			DispatchThrottlingRateInMsg:  int64(*params.DispatchRate.DispatchThrottlingRateInMsg),
+		var dispatchThrottlingRateInMsg int64
+		if params.DispatchRate.DispatchThrottlingRateInMsg != nil {
+			dispatchThrottlingRateInMsg = int64(*params.DispatchRate.DispatchThrottlingRateInMsg)
+		}
+		dispatchRateData := utils.DispatchRateData{
+			DispatchThrottlingRateInMsg:  dispatchThrottlingRateInMsg,

-			RatePeriodInSecond:           int64(*params.DispatchRate.RatePeriodInSecond),
+			RatePeriodInSecond: func() int64 {
+				if params.DispatchRate.RatePeriodInSecond != nil {
+					return int64(*params.DispatchRate.RatePeriodInSecond)
+				}
+				return 0 // Default value if RatePeriodInSecond is nil
+			}(),

-			PublishThrottlingRateInMsg:  int64(*params.PublishRate.PublishThrottlingRateInMsg),
+			PublishThrottlingRateInMsg: func() int64 {
+				if params.PublishRate.PublishThrottlingRateInMsg != nil {
+					return int64(*params.PublishRate.PublishThrottlingRateInMsg)
+				}
+				return 0
+			}(),

		PublishThrottlingRateInMsg: int64(*params.PublishRate.PublishThrottlingRateInMsg),
		PublishThrottlingRateInByte: *params.PublishRate.PublishThrottlingRateInByte,

feat: enhance Topic with missing policies #337

feat: enhance Topic with missing policies #337

Uh oh!

Conversation

freeznet commented Jul 25, 2025

Motivation

Modifications

Verifying this change

Documentation

Uh oh!

github-actions bot commented Jul 25, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants