feat: align topic level latest apis for policies #338
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…0013016-1f9e82805e54
…g subscribe rate, offload policies, auto subscription creation, and schema compatibility strategy
Contributor
|
@freeznet:Thanks for your contribution. For this PR, do we need to update docs? |
github-actions
bot
added
the
doc-info-missing
… including subscribeRate, schemaCompatibilityStrategy, and autoSubscriptionCreation
…mespace and PulsarTopic - Updated PulsarNamespaceSpec and PulsarTopicSpec to use local type definitions for SchemaCompatibilityStrategy, OffloadPolicies, and AutoSubscriptionCreationOverride. - Enhanced deepcopy functions for new local types to ensure proper Kubernetes deep copy generation. - Removed unnecessary imports of adminutils from relevant files. - Adjusted tests to reflect changes in type definitions for schema compatibility strategy and offload policies.
Contributor
There was a problem hiding this comment.
Pull Request Overview
This PR enhances the PulsarTopic CRD to align with Pulsar's latest admin APIs by adding comprehensive topic-level policy configurations. The changes enable full declarative management of Pulsar topics through Kubernetes resources, bringing feature parity with Pulsar's admin APIs.
Key changes include:
- Extended PulsarTopicSpec with 11 new policy fields for rate limiting, message constraints, schema management, and offload policies
- Migrated from external package types to local type definitions for better Kubernetes integration
- Updated admin client implementation to apply all new policies during topic creation/updates
Reviewed Changes
Copilot reviewed 12 out of 13 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| tests/utils/spec.go | Added utility functions for creating test topics with new policy configurations |
| tests/operator/resources_test.go | Added comprehensive integration tests for new topic policies |
| pkg/connection/reconcile_topic.go | Updated topic parameter mapping to include all new policy fields |
| pkg/admin/interface.go | Extended TopicParams interface with new policy parameters and updated type references |
| pkg/admin/impl.go | Implemented admin client methods for applying new policies and added type conversion functions |
| go.work.sum, go.mod | Updated Pulsar client dependency version |
| docs/pulsar_topic.md | Added documentation for all new policy fields with examples |
| config/crd/bases/resource.streamnative.io_pulsartopics.yaml | Updated CRD schema with new policy field definitions |
| api/v1alpha1/zz_generated.deepcopy.go | Updated generated deepcopy methods for new types |
| api/v1alpha1/pulsartopic_types.go | Added new policy fields and local type definitions |
| api/v1alpha1/pulsarnamespace_types.go | Updated to use local SchemaCompatibilityStrategy type |
Comments suppressed due to low confidence (4)
tests/operator/resources_test.go:928
- [nitpick] The import alias 'pointer' is inconsistent with the rest of the test file which uses 'ptr'. Consider using 'ptr.Toint32' to maintain consistency.
topic.Spec.SubscribeRate.SubscribeThrottlingRatePerConsumer = pointer.Int32(20)
tests/operator/resources_test.go:929
- [nitpick] The import alias 'pointer' is inconsistent with the rest of the test file which uses 'ptr'. Consider using 'ptr.Toint32' to maintain consistency.
topic.Spec.SubscribeRate.RatePeriodInSecond = pointer.Int32(60)
api/v1alpha1/zz_generated.deepcopy.go:22
- The import alias 'pkgutils' for the external library and 'utils' for the local package creates confusion. Consider using more descriptive aliases like 'pulsarutils' for the external library.
pkgutils "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
api/v1alpha1/zz_generated.deepcopy.go:23
- The import alias 'utils' for the local package creates potential confusion with the previous 'pkgutils'. Consider using 'localutils' or a more specific alias.
"github.com/streamnative/pulsar-resources-operator/pkg/utils"
Comment on lines
+450
to
+451
| SubscribeThrottlingRatePerConsumer: -1, // default to unlimited | ||
| RatePeriodInSecond: 30, // default period |
There was a problem hiding this comment.
The default values are hardcoded (-1 and 30). Consider defining these as named constants at the package level to improve maintainability and make the defaults explicit.
Suggested change
| SubscribeThrottlingRatePerConsumer: -1, // default to unlimited | |
| RatePeriodInSecond: 30, // default period | |
| SubscribeThrottlingRatePerConsumer: DefaultSubscribeThrottlingRatePerConsumer, | |
| RatePeriodInSecond: DefaultRatePeriodInSeconds, |
labuladong
previously approved these changes
Aug 1, 2025
maxsxu
previously approved these changes
Aug 1, 2025
…rces test - Updated the test for ManagedLedgerOffloadMaxThreads to directly compare with an integer instead of casting to int32, improving readability and consistency in the test code.
maxsxu
approved these changes
Aug 2, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
(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: #
feat: align topic level latest apis for policies
Master Issue: #
Motivation
The Pulsar Resources Operator's PulsarTopic CRD was missing several important topic-level policy configurations that are available in the latest Pulsar admin APIs. This gap prevented Kubernetes operators from fully managing Pulsar topics declaratively with all available policy options.
Users managing Pulsar clusters through Kubernetes need the ability to configure comprehensive topic policies including:
Without these policies in the CRD, operators had to resort to manual API calls or scripts to configure topics fully, breaking the declarative GitOps workflow. This PR brings feature parity with Pulsar's admin APIs, enabling full topic lifecycle management through Kubernetes resources.
Modifications
This PR adds comprehensive topic-level policy support to the PulsarTopic CRD by:
1. Extended PulsarTopicSpec with 11 new policy fields:
SubscribeRate- Control subscription throttling ratesMaxMessageSize- Set maximum message size in bytesMaxProducers- Limit number of producers per topicMaxConsumers- Limit number of consumers per topicDeduplication- Enable/disable message deduplicationPublishRate- Control publish rate limitingSchemaValidationEnforced- Enforce schema validationOffloadPolicies- Configure tiered storage offload behaviorAutoSubscriptionCreation- Control automatic subscription creationSchemaCompatibilityStrategy- Set schema evolution rulesReplicatorDispatchRate- Control replication dispatch rates2. Type System Improvements:
SchemaCompatibilityStrategy,OffloadPolicies, andAutoSubscriptionCreationOverride3. Admin API Integration:
TopicParamsinterface with all new policy parameters4. Resource Reconciliation:
reconcile_topic.goto map all new spec fields to admin parameters5. CRD Schema Updates:
Verifying this change
(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:)
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)