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")