From 754d4cb7f433adadd05b2a47485bc0bedfa8cdf8 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 14 Aug 2026 15:00:14 -0700 Subject: [PATCH] fix(namespace): converge bookie affinity group instead of blind-writing Every PulsarNamespace that did not declare `bookieAffinityGroup` issued an unconditional DELETE .../persistence/bookieAffinity on each reconcile, with no IsNotFound tolerance -- unlike the neighbouring RemoveTopicAutoCreation and RemoveInactiveTopicPolicies branches in the same function. Pulsar implements the delete as a set of a null group (NamespacesBase#internalDeleteBookieAffinityGroupAsync), and both the setter and the getter call validateSuperUserAccessAsync(). So a tenant-admin connection got 401/403 on that DELETE, ApplyNamespace failed, and the namespace never reached Ready -- for users who never touched bookie affinity at all. It also created the local-policies znode with default bundle data when there was nothing to delete. Read the current group first and write only on an actual diff. An unset-to-unset transition now touches Pulsar not at all, which removes the superuser dependency for namespaces that do not use the feature. Both of Pulsar's "no group" answers are normalized: 404 "Namespace local-policies does not exist" when the namespace has no local policies, and 200 with an empty group when they were cleared. Also: - Correct the BookieAffinityGroup godoc, which described PulsarNSIsolationPolicy -- a different feature -- and made the field read as something it is not. - Require MinLength=1 on both group names. An empty primary passed CRD validation and was pushed to Pulsar, where BookieRackAffinityMapping cannot place ledgers. - Document the end-to-end carve-out: PulsarNSIsolationPolicy for brokers paired with bookieAffinityGroup for bookies, including the rack-metadata prerequisite and the superuser requirement. The CRD schema was updated by hand rather than via `make manifests`: the Makefile pins controller-gen v0.17.0 but the committed CRDs were generated with v0.15.0, so regenerating rewrites all 19 CRDs and strips their license headers. The bookieAffinityGroup block matches the generator output exactly. The version skew is pre-existing and needs its own change. Fixes #420 Co-Authored-By: Claude Opus 5 (1M context) --- api/v1alpha1/pulsarnamespace_types.go | 15 +- ...urce.streamnative.io_pulsarnamespaces.yaml | 15 +- ...urce.streamnative.io_pulsarnamespaces.yaml | 15 +- docs/pulsar_namespace.md | 89 ++++++- docs/pulsar_ns_isolation_policy.md | 2 + pkg/admin/bookie_affinity_group_test.go | 217 ++++++++++++++++++ pkg/admin/impl.go | 73 ++++-- tests/operator/resources_test.go | 97 ++++++++ tests/utils/spec.go | 11 + 9 files changed, 515 insertions(+), 19 deletions(-) create mode 100644 pkg/admin/bookie_affinity_group_test.go diff --git a/api/v1alpha1/pulsarnamespace_types.go b/api/v1alpha1/pulsarnamespace_types.go index 4e9e9bca..0781d503 100644 --- a/api/v1alpha1/pulsarnamespace_types.go +++ b/api/v1alpha1/pulsarnamespace_types.go @@ -253,7 +253,12 @@ type PulsarNamespaceSpec struct { // +optional Deduplication *bool `json:"deduplication,omitempty"` - // BookieAffinityGroup is the name of the namespace isolation policy to apply to the namespace. + // BookieAffinityGroup pins the namespace's ledgers to the bookies that belong to the + // named BookKeeper rack/affinity groups. This is the storage half of a namespace + // carve-out; pair it with a PulsarNSIsolationPolicy to also pin the namespace to a + // dedicated set of brokers. + // Omitting this field removes any affinity group previously set for the namespace. + // +optional BookieAffinityGroup *BookieAffinityGroupData `json:"bookieAffinityGroup,omitempty"` // TopicAutoCreationConfig controls whether automatic topic creation is allowed in this namespace @@ -344,9 +349,17 @@ type PulsarNamespaceSpec struct { SchemaAutoUpdateCompatibilityStrategy *adminutils.SchemaAutoUpdateCompatibilityStrategy `json:"schemaAutoUpdateCompatibilityStrategy,omitempty"` } +// BookieAffinityGroupData selects the BookKeeper rack/affinity groups that a namespace's +// ledgers are placed on. The group names must match the rack metadata already registered +// for the bookies; the operator does not create group membership. type BookieAffinityGroupData struct { + // BookkeeperAffinityGroupPrimary is the group bookies are selected from first. + // +kubebuilder:validation:MinLength=1 BookkeeperAffinityGroupPrimary string `json:"bookkeeperAffinityGroupPrimary"` + // BookkeeperAffinityGroupSecondary is the group bookies are selected from when the + // primary group cannot satisfy the ensemble. + // +kubebuilder:validation:MinLength=1 // +optional BookkeeperAffinityGroupSecondary string `json:"bookkeeperAffinityGroupSecondary,omitempty"` } diff --git a/charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsarnamespaces.yaml b/charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsarnamespaces.yaml index 45bc1ae7..7b932106 100644 --- a/charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsarnamespaces.yaml +++ b/charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsarnamespaces.yaml @@ -113,12 +113,23 @@ spec: - message_age type: string bookieAffinityGroup: - description: BookieAffinityGroup is the name of the namespace isolation - policy to apply to the namespace. + description: |- + BookieAffinityGroup pins the namespace's ledgers to the bookies that belong to the + named BookKeeper rack/affinity groups. This is the storage half of a namespace + carve-out; pair it with a PulsarNSIsolationPolicy to also pin the namespace to a + dedicated set of brokers. + Omitting this field removes any affinity group previously set for the namespace. properties: bookkeeperAffinityGroupPrimary: + description: BookkeeperAffinityGroupPrimary is the group bookies + are selected from first. + minLength: 1 type: string bookkeeperAffinityGroupSecondary: + description: |- + BookkeeperAffinityGroupSecondary is the group bookies are selected from when the + primary group cannot satisfy the ensemble. + minLength: 1 type: string required: - bookkeeperAffinityGroupPrimary diff --git a/config/crd/bases/resource.streamnative.io_pulsarnamespaces.yaml b/config/crd/bases/resource.streamnative.io_pulsarnamespaces.yaml index 45bc1ae7..7b932106 100644 --- a/config/crd/bases/resource.streamnative.io_pulsarnamespaces.yaml +++ b/config/crd/bases/resource.streamnative.io_pulsarnamespaces.yaml @@ -113,12 +113,23 @@ spec: - message_age type: string bookieAffinityGroup: - description: BookieAffinityGroup is the name of the namespace isolation - policy to apply to the namespace. + description: |- + BookieAffinityGroup pins the namespace's ledgers to the bookies that belong to the + named BookKeeper rack/affinity groups. This is the storage half of a namespace + carve-out; pair it with a PulsarNSIsolationPolicy to also pin the namespace to a + dedicated set of brokers. + Omitting this field removes any affinity group previously set for the namespace. properties: bookkeeperAffinityGroupPrimary: + description: BookkeeperAffinityGroupPrimary is the group bookies + are selected from first. + minLength: 1 type: string bookkeeperAffinityGroupSecondary: + description: |- + BookkeeperAffinityGroupSecondary is the group bookies are selected from when the + primary group cannot satisfy the ensemble. + minLength: 1 type: string required: - bookkeeperAffinityGroupPrimary diff --git a/docs/pulsar_namespace.md b/docs/pulsar_namespace.md index d8d5612b..dc9cc215 100644 --- a/docs/pulsar_namespace.md +++ b/docs/pulsar_namespace.md @@ -28,7 +28,7 @@ The `PulsarNamespace` resource defines a namespace in a Pulsar cluster. It allow | `geoReplicationRefs` | List of references to PulsarGeoReplication resources, used to configure geo-replication for this namespace. Use only when using PulsarGeoReplication for setting up geo-replication between two Pulsar instances. | No | | `replicationClusters` | List of clusters to which the namespace is replicated. Use only if replicating clusters within the same Pulsar instance. | No | | `deduplication` | Whether to enable message deduplication for the namespace. | No | -| `bookieAffinityGroup` | Set the bookie-affinity group for the namespace, which has two sub fields: `bookkeeperAffinityGroupPrimary(String)` is required, and `bookkeeperAffinityGroupSecondary(String)` is optional. | No | +| `bookieAffinityGroup` | Pins the namespace's ledgers to the bookies belonging to the named BookKeeper rack/affinity groups. Two sub fields: `bookkeeperAffinityGroupPrimary(String)` is required, `bookkeeperAffinityGroupSecondary(String)` is optional. See [Broker and Bookie Isolation](#broker-and-bookie-isolation). | No | | `topicAutoCreationConfig` | Configures automatic topic creation behavior within this namespace. Contains settings for whether auto-creation is allowed, the type of topics created, and default number of partitions. | No | | `schemaCompatibilityStrategy` | Schema compatibility strategy for this namespace. Controls how schema evolution is handled for topics within this namespace. Options: `UNDEFINED`, `ALWAYS_INCOMPATIBLE`, `ALWAYS_COMPATIBLE`, `BACKWARD`, `FORWARD`, `FULL`, `BACKWARD_TRANSITIVE`, `FORWARD_TRANSITIVE`, `FULL_TRANSITIVE`. | No | | `schemaValidationEnforced` | Controls whether schema validation is enforced for this namespace. When enabled, producers must provide a schema when publishing messages. If not specified, the cluster's default schema validation enforcement setting will be used. | No | @@ -309,6 +309,93 @@ persistencePolicies: managedLedgerMaxMarkDeleteRate: "10.0" ``` +## Broker and Bookie Isolation + +A namespace carve-out on a shared cluster has two halves, and they are configured through +two different resources: + +| Half | Pins | Resource | +| --- | --- | --- | +| Brokers | Which brokers may own the namespace's bundles | [`PulsarNSIsolationPolicy`](pulsar_ns_isolation_policy.md) | +| Bookies | Which bookies the namespace's ledgers are written to | `PulsarNamespace.spec.bookieAffinityGroup` | + +The two are independent — configure either on its own, or both together for a full +"virtual cluster" per tenant or subsystem. + +### Bookie Affinity Groups + +`bookieAffinityGroup` is the declarative equivalent of +`pulsar-admin namespaces set-bookie-affinity-group`: + +```yaml +bookieAffinityGroup: + bookkeeperAffinityGroupPrimary: subsystem-a # required + bookkeeperAffinityGroupSecondary: subsystem-a-dr # optional +``` + +Bookies are selected from the primary group first, falling back to the secondary group when +the primary cannot satisfy the ensemble. Removing the `bookieAffinityGroup` field from the +CR removes the affinity policy from the namespace, returning it to cluster-wide bookie +placement. + +**Prerequisite:** the group names must match rack metadata that the bookies already carry +(`bookkeeper-rack-aware` placement, set through bookie configuration or BookKeeper metadata). +The operator selects among existing groups; it does not create group membership. Setting an +affinity group whose members cannot satisfy the namespace's ensemble size will cause writes +to fail, so verify group membership before applying. + +**Permissions:** Pulsar requires superuser access to read, set, or clear a namespace's bookie +affinity group. A `PulsarConnection` whose credentials are only tenant-admin can manage +namespaces normally, but cannot use this field. + +### Complete Carve-Out Example + +Pinning `finance/transactions` to dedicated brokers *and* dedicated bookies: + +```yaml +apiVersion: resource.streamnative.io/v1alpha1 +kind: PulsarNSIsolationPolicy +metadata: + name: finance-isolation + namespace: default +spec: + name: finance-isolation + cluster: my-pulsar-cluster + connectionRef: + name: my-connection + namespaces: + - finance/.* + primary: + - broker-finance-.*\.example\.com + secondary: + - broker-shared-.*\.example\.com + autoFailoverPolicyType: min_available + autoFailoverPolicyParams: + min_limit: "1" + usage_threshold: "80" +--- +apiVersion: resource.streamnative.io/v1alpha1 +kind: PulsarNamespace +metadata: + name: finance-transactions + namespace: default +spec: + name: finance/transactions + connectionRef: + name: my-connection + bookieAffinityGroup: + bookkeeperAffinityGroupPrimary: finance + bookkeeperAffinityGroupSecondary: shared + persistencePolicies: + bookkeeperEnsemble: 3 + bookkeeperWriteQuorum: 3 + bookkeeperAckQuorum: 2 +``` + +Note that `persistencePolicies` and `bookieAffinityGroup` work together: the affinity group +decides *which* bookies are eligible, and the ensemble/quorum settings decide *how many* of +them each ledger uses. The eligible groups must contain at least `bookkeeperEnsemble` bookies. + ## Topic Management Policies ### Compaction Configuration diff --git a/docs/pulsar_ns_isolation_policy.md b/docs/pulsar_ns_isolation_policy.md index 68e5a6db..3160951c 100644 --- a/docs/pulsar_ns_isolation_policy.md +++ b/docs/pulsar_ns_isolation_policy.md @@ -4,6 +4,8 @@ The `PulsarNSIsolationPolicy` resource defines a ns-isolation-policy in a Pulsar cluster. It allows you to configure namespace isolation policies to limit the set of brokers that can be used for assignment. +This resource covers the broker half of a namespace carve-out. To also pin a namespace's storage to a dedicated set of bookies, pair it with `bookieAffinityGroup` on the [`PulsarNamespace`](pulsar_namespace.md#broker-and-bookie-isolation) resource. + ## Specifications | Field | Description | Required | diff --git a/pkg/admin/bookie_affinity_group_test.go b/pkg/admin/bookie_affinity_group_test.go new file mode 100644 index 00000000..55d3fe81 --- /dev/null +++ b/pkg/admin/bookie_affinity_group_test.go @@ -0,0 +1,217 @@ +// Copyright 2026 StreamNative +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package admin + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + pulsaradmin "github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin" + "github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config" + + resourcev1alpha1 "github.com/streamnative/pulsar-resources-operator/api/v1alpha1" +) + +const bookieAffinityPath = "/admin/v2/namespaces/public/default/persistence/bookieAffinity" + +// currentAffinity models what Pulsar reports for a namespace's bookie affinity group. +type currentAffinity struct { + // status is the response code for GET, 404 meaning the namespace has no local policies. + status int + // body is the JSON payload returned when status is 200. + body string +} + +var ( + affinityAbsent = currentAffinity{status: http.StatusNotFound} + // Pulsar answers 200 with an empty group when local policies exist but the affinity + // group was cleared. + affinityCleared = currentAffinity{status: http.StatusOK, body: `{}`} + affinityGroupA = currentAffinity{ + status: http.StatusOK, + body: `{"bookkeeperAffinityGroupPrimary":"group-a","bookkeeperAffinityGroupSecondary":"group-b"}`, + } +) + +func TestApplyBookieAffinityGroup(t *testing.T) { + tests := []struct { + name string + desired *resourcev1alpha1.BookieAffinityGroupData + current currentAffinity + wantMethods []string + wantPostBody map[string]string + }{ + { + // Setting and clearing the affinity group both require superuser access in + // Pulsar, so a namespace that never uses the feature must not write at all. + name: "unset stays unset without writing", + desired: nil, + current: affinityAbsent, + wantMethods: []string{http.MethodGet}, + }, + { + name: "cleared group is not deleted again", + desired: nil, + current: affinityCleared, + wantMethods: []string{http.MethodGet}, + }, + { + name: "removing the setting deletes the group", + desired: nil, + current: affinityGroupA, + wantMethods: []string{http.MethodGet, http.MethodDelete}, + }, + { + name: "group is set when none exists", + desired: &resourcev1alpha1.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: "group-a", + BookkeeperAffinityGroupSecondary: "group-b", + }, + current: affinityAbsent, + wantMethods: []string{http.MethodGet, http.MethodPost}, + wantPostBody: map[string]string{ + "bookkeeperAffinityGroupPrimary": "group-a", + "bookkeeperAffinityGroupSecondary": "group-b", + }, + }, + { + name: "matching group is left alone", + desired: &resourcev1alpha1.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: "group-a", + BookkeeperAffinityGroupSecondary: "group-b", + }, + current: affinityGroupA, + wantMethods: []string{http.MethodGet}, + }, + { + name: "changed group is rewritten", + desired: &resourcev1alpha1.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: "group-c", + }, + current: affinityGroupA, + wantMethods: []string{http.MethodGet, http.MethodPost}, + wantPostBody: map[string]string{ + "bookkeeperAffinityGroupPrimary": "group-c", + "bookkeeperAffinityGroupSecondary": "", + }, + }, + { + // Dropping the secondary group is a real change even though the primary matches. + name: "dropping the secondary group is rewritten", + desired: &resourcev1alpha1.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: "group-a", + }, + current: affinityGroupA, + wantMethods: []string{http.MethodGet, http.MethodPost}, + wantPostBody: map[string]string{ + "bookkeeperAffinityGroupPrimary": "group-a", + "bookkeeperAffinityGroupSecondary": "", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var methods []string + var postBody map[string]string + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.EscapedPath() != bookieAffinityPath { + t.Errorf("unexpected request to %s", r.URL.EscapedPath()) + w.WriteHeader(http.StatusNotImplemented) + return + } + methods = append(methods, r.Method) + + switch r.Method { + case http.MethodGet: + if tt.current.status != http.StatusOK { + http.Error(w, "Namespace local-policies does not exist", tt.current.status) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, tt.current.body) + case http.MethodPost: + if err := json.NewDecoder(r.Body).Decode(&postBody); err != nil { + t.Errorf("decode bookie affinity payload: %v", err) + } + w.WriteHeader(http.StatusNoContent) + default: + w.WriteHeader(http.StatusNoContent) + } + })) + t.Cleanup(server.Close) + + client, err := pulsaradmin.New(&config.Config{WebServiceURL: server.URL}) + if err != nil { + t.Fatalf("create Pulsar admin client: %v", err) + } + + adminClient := &PulsarAdminClient{adminClient: client} + if err := adminClient.applyBookieAffinityGroup("public/default", tt.desired); err != nil { + t.Fatalf("apply bookie affinity group: %v", err) + } + + if len(methods) != len(tt.wantMethods) { + t.Fatalf("requests = %v, want %v", methods, tt.wantMethods) + } + for i := range tt.wantMethods { + if methods[i] != tt.wantMethods[i] { + t.Fatalf("request[%d] = %s, want %s", i, methods[i], tt.wantMethods[i]) + } + } + + if tt.wantPostBody == nil { + return + } + for key, want := range tt.wantPostBody { + if got := postBody[key]; got != want { + t.Fatalf("post body %q = %q, want %q", key, got, want) + } + } + }) + } +} + +// A GET failure that is not a 404 must surface rather than be mistaken for "no group set", +// which would otherwise let the operator silently skip a required write. +func TestApplyBookieAffinityGroupPropagatesReadFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Errorf("unexpected %s request after a failed read", r.Method) + } + http.Error(w, "Don't have admin permission", http.StatusForbidden) + })) + t.Cleanup(server.Close) + + client, err := pulsaradmin.New(&config.Config{WebServiceURL: server.URL}) + if err != nil { + t.Fatalf("create Pulsar admin client: %v", err) + } + + adminClient := &PulsarAdminClient{adminClient: client} + err = adminClient.applyBookieAffinityGroup("public/default", &resourcev1alpha1.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: "group-a", + }) + if err == nil { + t.Fatal("apply bookie affinity group succeeded, want error") + } + if reason := ErrorReason(err); reason != ReasonForbidden { + t.Fatalf("error reason = %v, want %v", reason, ReasonForbidden) + } +} diff --git a/pkg/admin/impl.go b/pkg/admin/impl.go index 260c08e9..16a6513a 100644 --- a/pkg/admin/impl.go +++ b/pkg/admin/impl.go @@ -1187,6 +1187,63 @@ func (p *PulsarAdminClient) RemoveTopicProperty(name string, persistent *bool, k return nil } +// getBookieAffinityGroup returns the bookie affinity group currently configured for the +// namespace, or nil when none is set. +// +// Pulsar reports "no affinity group" in two different ways: a 404 ("Namespace +// local-policies does not exist") when the namespace has no local policies at all, and a +// 200 carrying an empty group when local policies exist but the affinity group was +// cleared. Both are normalized to nil here. +func (p *PulsarAdminClient) getBookieAffinityGroup(completeNSName string) (*utils.BookieAffinityGroupData, error) { + current, err := p.adminClient.Namespaces().GetBookieAffinityGroup(completeNSName) + if err != nil { + if IsNotFound(err) { + return nil, nil + } + return nil, err + } + if current == nil || current.BookkeeperAffinityGroupPrimary == "" { + return nil, nil + } + return current, nil +} + +// applyBookieAffinityGroup moves the namespace's bookie affinity group towards the desired +// state, writing only when it actually differs from what Pulsar reports. +// +// Setting, clearing and reading the affinity group all require superuser access in Pulsar +// (NamespacesBase#internalSetBookieAffinityGroupAsync validates it, and the delete path is +// implemented as a set of a null group). Reconciling unconditionally would therefore make +// every namespace depend on superuser credentials, even ones that never use the feature, +// so an unset-to-unset transition must not touch Pulsar at all. +func (p *PulsarAdminClient) applyBookieAffinityGroup(completeNSName string, + desired *v1alpha1.BookieAffinityGroupData) error { + current, err := p.getBookieAffinityGroup(completeNSName) + if err != nil { + return err + } + + if desired == nil { + if current == nil { + return nil + } + if err := p.adminClient.Namespaces().DeleteBookieAffinityGroup(completeNSName); err != nil && + !IsNotFound(err) { + return err + } + return nil + } + + target := utils.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: desired.BookkeeperAffinityGroupPrimary, + BookkeeperAffinityGroupSecondary: desired.BookkeeperAffinityGroupSecondary, + } + if current != nil && *current == target { + return nil + } + return p.adminClient.Namespaces().SetBookieAffinityGroup(completeNSName, target) +} + func (p *PulsarAdminClient) applyNamespacePolicies(completeNSName string, params *NamespaceParams) error { naName, err := utils.GetNamespaceName(completeNSName) if err != nil { @@ -1377,19 +1434,9 @@ func (p *PulsarAdminClient) applyNamespacePolicies(completeNSName string, params // The pulsar-client-go library doesn't have DeletePersistence for namespaces, // and sending empty PersistencePolicies{} with BookkeeperEnsemble=0 causes // validation errors (Bookkeeper-Ensemble must be > 0 and <= 5). - if params.BookieAffinityGroup != nil { - err = p.adminClient.Namespaces().SetBookieAffinityGroup(completeNSName, utils.BookieAffinityGroupData{ - BookkeeperAffinityGroupPrimary: params.BookieAffinityGroup.BookkeeperAffinityGroupPrimary, - BookkeeperAffinityGroupSecondary: params.BookieAffinityGroup.BookkeeperAffinityGroupSecondary, - }) - if err != nil { - return err - } - } else { - err = p.adminClient.Namespaces().DeleteBookieAffinityGroup(completeNSName) - if err != nil { - return err - } + + if err := p.applyBookieAffinityGroup(completeNSName, params.BookieAffinityGroup); err != nil { + return err } // Handle topic auto-creation configuration diff --git a/tests/operator/resources_test.go b/tests/operator/resources_test.go index 8b4ac239..af421968 100644 --- a/tests/operator/resources_test.go +++ b/tests/operator/resources_test.go @@ -1200,6 +1200,103 @@ var _ = Describe("Resources", func() { }) }) + Context("PulsarNamespace Bookie Affinity Group", Ordered, func() { + var ( + bookieAffinityNamespace *v1alphav1.PulsarNamespace + bookieAffinityNamespaceName string = "test-bookie-affinity-namespace" + ) + + BeforeAll(func() { + bookieAffinityNamespace = utils.MakePulsarNamespaceWithBookieAffinityGroup( + namespaceName, + bookieAffinityNamespaceName, + "cloud/bookie-affinity", + pconnName, + lifecyclePolicy, + ) + }) + + It("should create namespace with bookie affinity group successfully", func() { + err := k8sClient.Create(ctx, bookieAffinityNamespace) + Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue()) + }) + + It("should be ready", func() { + Eventually(func() bool { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(ns) + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should have correct bookie affinity group configuration", func() { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + + Expect(ns.Spec.BookieAffinityGroup).ShouldNot(BeNil()) + Expect(ns.Spec.BookieAffinityGroup.BookkeeperAffinityGroupPrimary).Should(Equal("group-primary")) + Expect(ns.Spec.BookieAffinityGroup.BookkeeperAffinityGroupSecondary).Should(Equal("group-secondary")) + }) + + It("should update the bookie affinity group successfully", func() { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + + ns.Spec.BookieAffinityGroup.BookkeeperAffinityGroupPrimary = "group-primary-updated" + ns.Spec.BookieAffinityGroup.BookkeeperAffinityGroupSecondary = "" + Expect(k8sClient.Update(ctx, ns)).Should(Succeed()) + }) + + It("should be ready after update", func() { + Eventually(func() bool { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(ns) && + ns.Status.ObservedGeneration == ns.Generation + }, "20s", "100ms").Should(BeTrue()) + }) + + It("should reject an empty primary affinity group", func() { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + + ns.Spec.BookieAffinityGroup.BookkeeperAffinityGroupPrimary = "" + Expect(k8sClient.Update(ctx, ns)).ShouldNot(Succeed()) + }) + + It("should remove the affinity policy when the field is dropped", func() { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + + ns.Spec.BookieAffinityGroup = nil + Expect(k8sClient.Update(ctx, ns)).Should(Succeed()) + + Eventually(func() bool { + ns := &v1alphav1.PulsarNamespace{} + Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + return v1alphav1.IsPulsarResourceReady(ns) && + ns.Status.ObservedGeneration == ns.Generation + }, "20s", "100ms").Should(BeTrue()) + }) + + AfterAll(func() { + if bookieAffinityNamespace != nil { + Eventually(func(g Gomega) { + ns := &v1alphav1.PulsarNamespace{} + tns := types.NamespacedName{Namespace: namespaceName, Name: bookieAffinityNamespaceName} + g.Expect(k8sClient.Get(ctx, tns, ns)).Should(Succeed()) + g.Expect(k8sClient.Delete(ctx, ns)).Should(Succeed()) + }).Should(Succeed()) + } + }) + }) + Context("PulsarTopic Auto Subscription Creation", Ordered, func() { var ( autoSubscriptionTopic *v1alphav1.PulsarTopic diff --git a/tests/utils/spec.go b/tests/utils/spec.go index 00e7ad82..0aa137b3 100644 --- a/tests/utils/spec.go +++ b/tests/utils/spec.go @@ -112,6 +112,17 @@ func MakePulsarNamespaceWithOffloadPolicies(namespace, name, namespaceName, conn return ns } +// MakePulsarNamespaceWithBookieAffinityGroup will generate a PulsarNamespace pinned to a +// pair of BookKeeper affinity groups +func MakePulsarNamespaceWithBookieAffinityGroup(namespace, name, namespaceName, connectionName string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarNamespace { + ns := MakePulsarNamespace(namespace, name, namespaceName, connectionName, policy) + ns.Spec.BookieAffinityGroup = &v1alpha1.BookieAffinityGroupData{ + BookkeeperAffinityGroupPrimary: "group-primary", + BookkeeperAffinityGroupSecondary: "group-secondary", + } + return ns +} + // MakePulsarNamespaceWithRateLimiting will generate a PulsarNamespace with rate limiting configurations func MakePulsarNamespaceWithRateLimiting(namespace, name, namespaceName, connectionName string, policy v1alpha1.PulsarResourceLifeCyclePolicy) *v1alpha1.PulsarNamespace { backlogSize := resource.MustParse("5Gi")