[PR] [feat][pip] PIP-470: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["wolfstudy (via GitHub)" <gi...@apache.org>]

wolfstudy opened a new pull request, #25774:
URL: https://github.com/apache/pulsar/pull/25774

   ### Motivation
   
   Pulsar brokers are statically bound to a single Bookie cluster via
   `bookkeeperMetadataServiceUri` at startup. When the underlying Bookie cluster
   hits capacity limits or non-recoverable failures, there is no runtime way to
   re-target brokers to another Bookie cluster without restart and full data
   migration. This PIP introduces zero-downtime Bookie cluster switching by
   attributing each ledger to its owning cluster and routing reads accordingly.
   
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["wolfstudy (via GitHub)" <gi...@apache.org>]

wolfstudy commented on PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#issuecomment-4514455974

   @lhotari @merlimat PTAL again, thanks
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["void-ptr974 (via GitHub)" <gi...@apache.org>]

void-ptr974 commented on code in PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#discussion_r3293940003


##########
pip/pip-477.md:
##########
@@ -0,0 +1,991 @@
+# PIP-477: Dynamic Bookie Cluster Switching for Brokers
+
+> **Revision 2 (Plugin-First Architecture)** — This revision reframes PIP-477
+> per the design feedback to build on top of the storage-layer pluggability
+> introduced by [PIP-384](./pip-384.md), the per-ledger metadata extension slot
+> introduced by [PIP-404](./pip-404.md), the `CustomCommandFactory` SPI introduced
+> by [PIP-201](./pip-201.md), and the existing `AdditionalServlet` SPI — instead
+> of adding a self-contained subsystem to Pulsar core.
+
+## Motivation
+
+Apache Pulsar's broker layer is bound to a fixed BookKeeper (Bookie) cluster
+at startup through the `bookkeeperMetadataServiceUri` configuration. This
+static binding becomes a bottleneck in several real-world operational
+scenarios:
+
+1. **Slow failure recovery** — When a Bookie cluster experiences capacity
+   exhaustion, sustained latency degradation, or non-recoverable failures, the
+   only mitigation today is to wait for the Bookie cluster to recover. In
+   elastic-storage deployments where one Bookie cluster backs 100+ broker
+   clusters, the blast radius is enormous.
+2. **No capacity escape hatch** — When a Bookie cluster approaches its
+   capacity ceiling, brokers cannot "spill over" to a fresh Bookie cluster.
+3. **No runtime re-targeting** — There is no supported runtime mechanism to
+   migrate brokers from one Bookie cluster to another without restart and full
+   data migration.
+4. **Data plane coupling** — Switching deals with the data plane: terabytes of
+   ledger bytes, ongoing publishes, ongoing acknowledgments, and live cursors.
+   A copy-everything-then-cutover approach is infeasible.
+
+This PIP proposes a **safe, ledger-attribution-driven framework** that allows
+operators to switch a broker cluster's underlying Bookie cluster at runtime,
+with **zero data-plane downtime** and **per-ledger routing** that preserves the
+readability of historical data on the old cluster.
+
+The framework enables:
+
+- **Zero-downtime switching** — publish and consume continue without interruption
+- **Per-ledger routing** — each ledger carries its own cluster attribution
+- **No bulk data migration** — Topic Data Ledgers age out naturally via TTL/Retention
+- **Targeted migration of small, long-lived metadata ledgers** — only Schema and
+  Cursor Ledgers are copied, preserving the original `ledgerId`
+- **Automatic rollback** within a configurable window
+- **Operator-driven, not auto-failover** — avoids split-brain
+
+## Design Principle: Build on PIP-384, Not Around It
+
+The feature is delivered as **a `ManagedLedgerStorage` implementation that
+holds multiple `BookkeeperManagedLedgerStorageClass` instances**, packaged as
+an out-of-tree NAR plugin. The framing PIP-384 establishes —
+*"a topic resolves to one storage class"* — is generalized here to:
+
+> *"A topic resolves to one **active** storage class at a time, with
+> **attribution-driven routing for previously-written ledgers**."*
+
+Concretely, this PIP:
+
+- Lives in a separate Maven module (`pulsar-bookie-cluster-switching`) and
+  ships as a NAR; the broker loads it via the existing `managedLedgerStorageClassName`
+  configuration hook (PIP-384).
+- Re-uses **`LedgerInfo.properties`** (PIP-404, tag 6) and
+  **`ManagedCursorInfo.cursorProperties`** (existing, tag 8) for per-ledger
+  cluster attribution. **Zero new proto fields** in `MLDataFormats.proto`.
+- Ships its REST surface via **`AdditionalServlet`**; ships its CLI via
+  **`CustomCommandFactory` NAR** (PIP-201). **Zero changes** to `pulsar-broker`
+  REST classes or `pulsar-admin` CLI core.
+- Requires only **three narrow SPI hooks** in core (Section "Required Core SPI
+  Additions"), each useful to *any* custom `ManagedLedgerStorage` author and
+  much smaller than inlining the full feature.
+- When the plugin NAR is not installed, broker behavior is **byte-for-byte
+  identical** to unmodified Pulsar; the ~99% of users that never switch BK
+  clusters pay zero cost in core surface area.
+
+## Goal
+
+Provide a runtime mechanism by which a broker cluster can be re-pointed from
+one Bookie cluster (`oldCluster`) to a new Bookie cluster (`newCluster`)
+without service interruption, while preserving the ability to read historical
+data that physically resides on `oldCluster`.
+
+### In Scope
+
+- A plugin-provided `ManagedLedgerStorage` implementation
+  (`MultiClusterManagedLedgerStorage`) that holds a
+  `Map<clusterName, BookkeeperManagedLedgerStorageClass>` and exposes the
+  currently active one.
+- Per-ledger cluster attribution stored in the **existing**
+  `LedgerInfo.properties` and `ManagedCursorInfo.cursorProperties` slots under
+  the reserved key `_pulsar.bookieClusterName`.
+- Per-broker-cluster registry of multiple Bookie clusters, stored in
+  Broker-ZK under a plugin-owned path.
+- A plugin-owned `BookieClusterReadRouter` that selects a BK client at
+  ledger-handle open time based on the attribution property.
+- A plugin-owned `BookieClusterSwitchOrchestrator` that drives
+  `BUILD → PROMOTE → CLEANUP` for Schema and Cursor Ledgers (migration
+  preserves the original `ledgerId` via `BookKeeper.asyncCreateLedgerAdv`).
+- Plugin-provided CLI (`pulsar-admin bookie-clusters …`) via PIP-201
+  `CustomCommandFactory`.
+- Plugin-provided REST API under `/admin/v2/bookie-clusters/*` via
+  `AdditionalServlet`.
+- Three narrow core SPI hooks (see Section "Required Core SPI Additions").
+
+### Out of Scope
+
+- **Topic Data Ledger migration** — not migrated; ages out via TTL/Retention.
+- **Automatic failure detection and switch triggering** — operator-initiated
+  to avoid split-brain.
+- **Cross-cluster (geo-replicated) coordination** — each broker cluster
+  switches independently.
+- **BookKeeper-cluster-side internal migration tools** — delegated to BookKeeper.
+- **Changing `metadataStoreUrl` (Broker-ZK)** — only the Bookie cluster is
+  switched; Broker-ZK remains the same throughout.
+- **New proto fields in `MLDataFormats.proto`** — reuses existing extension
+  slots from PIP-404 and `cursorProperties`.
+
+## High-Level Architecture
+
+```
+┌────────────────────────────────────────────────────────────────────────┐
+│ Pulsar core (unchanged + 3 narrow SPI hooks)                           │
+│                                                                         │
+│   ManagedLedgerStorage SPI            ─── PIP-384                       │
+│   ManagedLedgerStorageClass SPI       ─── PIP-384                       │
+│   LedgerInfo.properties (KV)          ─── PIP-404                       │
+│   ManagedCursorInfo.cursorProperties  ─── pre-existing                  │
+│   AdditionalServlet SPI               ─── pre-existing                  │
+│   CustomCommandFactory SPI            ─── PIP-201                       │
+│                                                                         │
+│   + ManagedLedgerConfig.activeBookKeeperSupplier         ←── NEW (S1)   │
+│   + ManagedLedgerConfig.bookKeeperResolver               ←── NEW (S2)   │
+│   + SchemaStorageBookKeeperProvider SPI                  ←── NEW (S3)   │
+└────────────────────────────┬───────────────────────────────────────────┘
+                             │ SPI / NAR loading
+┌────────────────────────────▼───────────────────────────────────────────┐
+│ pulsar-bookie-cluster-switching (NAR, opt-in)                          │
+│                                                                         │
+│   MultiClusterManagedLedgerStorage   impl ManagedLedgerStorage         │
+│     ├─ Map<name, BookkeeperManagedLedgerStorageClass>                  │
+│     ├─ activeClusterName (volatile)                                     │
+│     └─ getDefaultStorageClass() → active                               │
+│                                                                         │
+│   RoutingManagedLedgerFactory        wraps ManagedLedgerFactoryImpl    │
+│     └─ Per-ledger BK resolution from LedgerInfo.properties             │
+│                                                                         │
+│   RoutingSchemaBookKeeperProvider    impl SchemaStorageBookKeeperProvider│
+│     └─ Per-position BK resolution from SchemaLocator extension         │
+│                                                                         │
+│   BookieClusterConfigManager / ConfigWatcher                            │
+│     └─ Owns ZK path /admin/bookie-clusters/* (plugin-owned)            │
+│                                                                         │
+│   BookieClusterSwitchOrchestrator    BUILD → PROMOTE → CLEANUP          │
+│                                                                         │
+│   REST  /admin/v2/bookie-clusters/*  via AdditionalServlet              │
+│   CLI   pulsar-admin bookie-clusters via CustomCommandFactory NAR       │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Key Principles
+
+| Principle | Description |
+|-----------|-------------|
+| **Attribution as Single Source of Truth** | Every ledger at creation is "stamped" with the active cluster name inside its existing metadata properties. Reads are routed by that stamp. No history table, no boundary heuristics, no side znode. |
+| **Reuse Extension Slots, Don't Add Proto Fields** | `LedgerInfo.properties` (PIP-404) and `ManagedCursorInfo.cursorProperties` carry the attribution. Only `SchemaStorageFormat.proto` needs one tiny extension (Section "Required Core SPI Additions" S3). |
+| **Topic Data: Don't Migrate** | Topic Data Ledgers have TTL/Retention; let them age out. Switching does not move terabytes of data. |
+| **Schema / Cursor: Migrate, but Preserve `ledgerId`** | We use `asyncCreateLedgerAdv(ledgerId, …)` to recreate the **same `ledgerId`** in the new cluster, copy entries with **identical `entryId`**, then CAS only the attribution property. Business keys (`cursorsLedgerId`, `position.ledgerId`, `position.entryId`) are byte-for-byte unchanged in Broker-ZK. |
+| **Plugin-First, Opt-In** | When the NAR is absent or `managedLedgerStorageClassName` ≠ the multi-cluster class, the broker is byte-for-byte unmodified Pulsar. |
+| **Operator-Driven, Single Active Cluster** | At any moment exactly one cluster is `status=ACTIVE` in Broker-ZK. Auto-failover is rejected. |
+| **Broker-ZK is the Switch Source-of-Truth** | All cluster registrations and the switch directive live in Broker-ZK under a plugin-owned path. No Global-ZK, no external coordinator. |
+
+## Required Core SPI Additions
+
+These are the **only** changes to Pulsar core required by this PIP. Each is a
+narrow, generally-useful hook that any custom `ManagedLedgerStorage`
+implementation could leverage — not specific to BK switching.
+
+### S1. `ManagedLedgerConfig.activeBookKeeperSupplier`
+
+**Problem.** Today `ManagedLedgerImpl` captures the `BookKeeper` client once
+in its constructor (via `ManagedLedgerFactoryImpl.bookkeeperFactory`). When a
+custom storage class needs to hot-swap the underlying BK client (e.g. cluster
+switch), there is no clean injection point on the write path.
+
+**Hook.** Add an optional supplier:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Supplier<BookKeeper> activeBookKeeperSupplier;   // nullable
+
+public Supplier<BookKeeper> getActiveBookKeeperSupplier() { … }
+public ManagedLedgerConfig setActiveBookKeeperSupplier(Supplier<BookKeeper> s) { … }
+```
+
+`ManagedLedgerImpl.createLedgerAfterClosed`, `rollCurrentLedgerIfFull`, and
+`ManagedCursorImpl.doCreateNewMetadataLedger` resolve the BK client as:
+
+```java
+BookKeeper bk = config.getActiveBookKeeperSupplier() != null
+        ? config.getActiveBookKeeperSupplier().get()
+        : this.bookKeeper;     // backward-compatible default
+```
+
+**Scope.** Pure additive. When the supplier is unset (default), behavior is
+identical. ~30 lines of change in `managed-ledger/`.
+
+### S2. `ManagedLedgerConfig.bookKeeperResolver`
+
+**Problem.** Reads (and deletes) need a *per-ledger* resolution path. Today
+`ManagedLedgerImpl` uses the constructor-captured `bookKeeper` for every
+`asyncOpenLedger`/`asyncDeleteLedger`. After a switch, ledgers that physically
+live on the old cluster must be opened with the old client.
+
+**Hook.** Add an optional resolver function keyed on `LedgerInfo`:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Function<LedgerInfo, BookKeeper> bookKeeperResolver;   // nullable
+
+// And the analogous function for cursor metadata ledgers:
+private Function<ManagedCursorInfo, BookKeeper> cursorBookKeeperResolver;
+```
+
+All call sites that today reference `this.bookKeeper` to open or delete a
+specific ledger (the comprehensive list is in
+`ManagedLedgerImpl.getLedgerHandle`, `internalAsyncOpenCursor`,
+`asyncDeleteLedger`, `asyncDeleteFromBookKeeper`, …) become:
+
+```java
+BookKeeper bk = resolveBookKeeperForLedger(ledgerInfo);
+```
+
+Where:
+
+```java
+private BookKeeper resolveBookKeeperForLedger(LedgerInfo li) {
+    var resolver = config.getBookKeeperResolver();
+    return resolver != null ? resolver.apply(li) : this.bookKeeper;
+}
+```
+
+**Scope.** Pure additive; when resolvers are unset, behavior is identical.
+~80 lines of change in `managed-ledger/`, mostly mechanical substitutions.
+
+### S3. `SchemaStorageBookKeeperProvider` SPI
+
+**Problem.** `BookkeeperSchemaStorage` captures `this.bookKeeper` at
+`start()` time (line ~101). It directly opens schema ledgers via that single
+client. To support per-ledger routing we need to inject a resolver and to
+attribute each schema ledger to a cluster.
+
+**Hook (two parts).**
+
+1. **Add a tiny extension slot to `SchemaStorageFormat.proto`** — this is the
+   *only* proto change in this PIP:
+
+   ```protobuf
+   message PositionInfo {
+       required int64 ledgerId = 1;
+       required int64 entryId = 2;
+       repeated KeyValue properties = 3;   // ★ NEW — generic extension slot,
+                                           // mirrors PIP-404 LedgerInfo.properties
+   }
+
+   message KeyValue {
+       required string key = 1;
+       required string value = 2;
+   }
+   ```
+
+   This is the minimal possible change: a generic `properties` slot symmetric
+   to PIP-404. It does **not** mention bookie clusters; any future plugin can
+   use it.
+
+2. **Add a provider SPI for the schema storage BK client:**
+
+   ```java
+   // pulsar-broker/.../service/schema/SchemaStorageBookKeeperProvider.java
+   public interface SchemaStorageBookKeeperProvider {
+       /** Returns the BK client to use for the given schema position. */
+       BookKeeper resolve(PositionInfo position);
+       /** Returns the BK client to use for *creating* a new schema ledger. */
+       BookKeeper active();
+   }
+   ```
+
+   `BookkeeperSchemaStorage` resolves its provider as:
+
+   ```java
+   var provider = pulsar.getSchemaStorageBookKeeperProvider();   // nullable
+   BookKeeper bk = provider != null ? provider.resolve(position) : this.bookKeeper;
+   ```
+
+   The default (provider == null) preserves today's behavior exactly.
+
+**Scope.** One proto field (additive, `repeated` is forward/backward
+compatible in proto2) + one SPI interface + ~50 lines in
+`BookkeeperSchemaStorage`. No new REST endpoint, no CLI change.
+
+### Summary: Core Diff Footprint
+
+| Component | LOC | Risk |
+|-----------|-----|------|
+| `ManagedLedgerConfig` (S1, S2 setters/getters) | ~40 | Trivial additive |
+| `ManagedLedgerImpl` / `ManagedCursorImpl` resolver wiring | ~80 | Mechanical, well-scoped |
+| `SchemaStorageFormat.proto` `PositionInfo.properties` (S3) | 1 field | Additive, proto2-safe |
+| `SchemaStorageBookKeeperProvider` SPI + `BookkeeperSchemaStorage` wiring | ~50 | Additive |
+| **Total core diff** | **~170 LOC, 1 new optional proto field** | **Zero behavioral change when SPI unused** |
+
+Everything else lives in the plugin NAR.
+
+## Plugin Module: `pulsar-bookie-cluster-switching`
+
+A new top-level Maven module that builds a NAR (`pulsar-bookie-cluster-switching-<version>.nar`).
+
+### Module Layout
+
+```
+pulsar-bookie-cluster-switching/
+├─ src/main/java/.../
+│   ├─ storage/
+│   │   ├─ MultiClusterManagedLedgerStorage.java    (implements ManagedLedgerStorage)
+│   │   ├─ MultiClusterStorageClass.java            (implements BookkeeperManagedLedgerStorageClass)
+│   │   └─ RoutingManagedLedgerFactory.java
+│   ├─ schema/
+│   │   └─ RoutingSchemaBookKeeperProvider.java     (implements SchemaStorageBookKeeperProvider)
+│   ├─ registry/
+│   │   ├─ BookieClusterConfigManager.java
+│   │   └─ BookieClusterConfigWatcher.java
+│   ├─ orchestrator/
+│   │   ├─ BookieClusterSwitchOrchestrator.java
+│   │   ├─ LedgerCopyUtil.java
+│   │   └─ IdgenPrecheckService.java
+│   ├─ rest/
+│   │   └─ BookieClusterAdminServlet.java           (implements AdditionalServlet)
+│   └─ cli/
+│       └─ BookieClusterCommandFactory.java         (implements CustomCommandFactory)
+└─ src/main/resources/META-INF/services/
+    ├─ org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet
+    └─ org.apache.pulsar.admin.cli.extensions.CustomCommandFactory
+```
+
+### Wiring on Broker Start
+
+The plugin is activated by setting in `broker.conf`:
+
+```properties
+managedLedgerStorageClassName=org.apache.pulsar.ext.bookieswitch.MultiClusterManagedLedgerStorage
+additionalServlets=bookie-clusters-admin
+additionalServletDirectory=./plugins
+```
+
+Then `ManagedLedgerStorage.create()` (PIP-384 entry point) reflectively loads
+`MultiClusterManagedLedgerStorage`. Inside its `initialize()`:
+
+1. Reads its own configuration namespace (prefix `bookieClusterSwitch.*`).
+2. Reads `/admin/bookie-clusters/*` from Broker-ZK to discover registered
+   clusters and resolve the initial `activeClusterName` (tier-1 →
+   Broker-ZK `status=ACTIVE` cluster; tier-2 → optional
+   `bookieClusterSwitch.currentClusterName` hint; tier-3 → derive from
+   `bookkeeperMetadataServiceUri` and auto-register as `ACTIVE`).
+3. Builds one `BookkeeperManagedLedgerStorageClass` per registered cluster
+   (each owns its own `BookKeeper` client and `StatsProvider`).
+4. Constructs a `RoutingManagedLedgerFactory` that decorates a base
+   `ManagedLedgerFactoryImpl` and injects the per-ledger BK resolver via the
+   S1/S2 hooks above.
+5. Starts `BookieClusterConfigWatcher` to receive switch directives.
+6. (Origin-broker-only after a switch) starts the orchestrator.
+
+The plugin's `AdditionalServlet` instance is independently discovered by
+`AdditionalServlets.load(…)` in `WebService`; the plugin's
+`CustomCommandFactory` is independently discovered by `pulsar-admin`'s NAR
+class loader.
+
+## Detailed Design
+
+### Per-Ledger Attribution Encoding
+
+| Ledger type | proto container | Encoding | Key |
+|-------------|-----------------|----------|-----|
+| Topic Data | `ManagedLedgerInfo.LedgerInfo.properties` (PIP-404, tag 6) | `KeyValue` | `_pulsar.bookieClusterName` |
+| Cursor | `ManagedCursorInfo.cursorProperties` (existing, tag 8) | `StringProperty` | `_pulsar.bookieClusterName` |
+| Schema | `SchemaStorageFormat.PositionInfo.properties` (NEW, tag 3, S3) | `KeyValue` | `_pulsar.bookieClusterName` |
+
+The reserved key prefix `_pulsar.` is **owned by Pulsar core** to avoid
+collisions with plugin- or user-defined properties. Plugin code uses a
+constant:
+
+```java
+public static final String ATTR_KEY = "_pulsar.bookieClusterName";
+```
+
+**Compatibility.**
+
+- **proto2 forward-compat:** `repeated KeyValue properties` already exists in
+  `LedgerInfo` (PIP-404). For `PositionInfo` it's added with a new tag (3),
+  matching the symmetric pattern.
+- **Old broker reading new metadata:** ignores `properties` it doesn't
+  understand; preserves them via `UnknownFieldSet` on rewrite (already the
+  case for PIP-404).
+- **New broker reading old metadata (no attribution):** the plugin's resolver
+  falls back to the active cluster (which equals the original cluster for
+  pre-switch ledgers; correct).
+
+### Write-Path Stamping
+
+`RoutingManagedLedgerFactory` wraps the base factory. When `ManagedLedger`
+asks for a new ledger:
+
+```java
+// Pseudocode of the stamping path
+String activeCluster = multiClusterStorage.getActiveClusterName();
+BookKeeper bk = multiClusterStorage.getBookKeeper(activeCluster);    // via S1
+LedgerHandle lh = bk.asyncCreateLedger(...);
+
+// In the same metadata CAS that records the new LedgerInfo:

Review Comment:
   It looks like the intended direction is to write the attribution in the same metadata update that first records the new `LedgerInfo`. Could we make the required core change for that explicit?
   
   Today `asyncAddLedgerProperty` is an after-the-fact update. To make this atomic, it seems `ManagedLedgerImpl` needs a creation-time way for the storage implementation to provide `LedgerInfo.properties` before calling `store.asyncUpdateLedgerIds(...)`. The cursor and schema paths seem to need the analogous behavior for `ManagedCursorInfo.cursorProperties` and `SchemaStorageFormat.PositionInfo.properties`.
   
   Could the PIP clarify whether this is part of the proposed SPI changes, and how legacy ledgers without attribution are distinguished from newly-created ledgers that should always have attribution?



##########
pip/pip-477.md:
##########
@@ -0,0 +1,991 @@
+# PIP-477: Dynamic Bookie Cluster Switching for Brokers
+
+> **Revision 2 (Plugin-First Architecture)** — This revision reframes PIP-477
+> per the design feedback to build on top of the storage-layer pluggability
+> introduced by [PIP-384](./pip-384.md), the per-ledger metadata extension slot
+> introduced by [PIP-404](./pip-404.md), the `CustomCommandFactory` SPI introduced
+> by [PIP-201](./pip-201.md), and the existing `AdditionalServlet` SPI — instead
+> of adding a self-contained subsystem to Pulsar core.
+
+## Motivation
+
+Apache Pulsar's broker layer is bound to a fixed BookKeeper (Bookie) cluster
+at startup through the `bookkeeperMetadataServiceUri` configuration. This
+static binding becomes a bottleneck in several real-world operational
+scenarios:
+
+1. **Slow failure recovery** — When a Bookie cluster experiences capacity
+   exhaustion, sustained latency degradation, or non-recoverable failures, the
+   only mitigation today is to wait for the Bookie cluster to recover. In
+   elastic-storage deployments where one Bookie cluster backs 100+ broker
+   clusters, the blast radius is enormous.
+2. **No capacity escape hatch** — When a Bookie cluster approaches its
+   capacity ceiling, brokers cannot "spill over" to a fresh Bookie cluster.
+3. **No runtime re-targeting** — There is no supported runtime mechanism to
+   migrate brokers from one Bookie cluster to another without restart and full
+   data migration.
+4. **Data plane coupling** — Switching deals with the data plane: terabytes of
+   ledger bytes, ongoing publishes, ongoing acknowledgments, and live cursors.
+   A copy-everything-then-cutover approach is infeasible.
+
+This PIP proposes a **safe, ledger-attribution-driven framework** that allows
+operators to switch a broker cluster's underlying Bookie cluster at runtime,
+with **zero data-plane downtime** and **per-ledger routing** that preserves the
+readability of historical data on the old cluster.
+
+The framework enables:
+
+- **Zero-downtime switching** — publish and consume continue without interruption
+- **Per-ledger routing** — each ledger carries its own cluster attribution
+- **No bulk data migration** — Topic Data Ledgers age out naturally via TTL/Retention
+- **Targeted migration of small, long-lived metadata ledgers** — only Schema and
+  Cursor Ledgers are copied, preserving the original `ledgerId`
+- **Automatic rollback** within a configurable window
+- **Operator-driven, not auto-failover** — avoids split-brain
+
+## Design Principle: Build on PIP-384, Not Around It
+
+The feature is delivered as **a `ManagedLedgerStorage` implementation that
+holds multiple `BookkeeperManagedLedgerStorageClass` instances**, packaged as
+an out-of-tree NAR plugin. The framing PIP-384 establishes —
+*"a topic resolves to one storage class"* — is generalized here to:
+
+> *"A topic resolves to one **active** storage class at a time, with
+> **attribution-driven routing for previously-written ledgers**."*
+
+Concretely, this PIP:
+
+- Lives in a separate Maven module (`pulsar-bookie-cluster-switching`) and
+  ships as a NAR; the broker loads it via the existing `managedLedgerStorageClassName`
+  configuration hook (PIP-384).
+- Re-uses **`LedgerInfo.properties`** (PIP-404, tag 6) and
+  **`ManagedCursorInfo.cursorProperties`** (existing, tag 8) for per-ledger
+  cluster attribution. **Zero new proto fields** in `MLDataFormats.proto`.
+- Ships its REST surface via **`AdditionalServlet`**; ships its CLI via
+  **`CustomCommandFactory` NAR** (PIP-201). **Zero changes** to `pulsar-broker`
+  REST classes or `pulsar-admin` CLI core.
+- Requires only **three narrow SPI hooks** in core (Section "Required Core SPI
+  Additions"), each useful to *any* custom `ManagedLedgerStorage` author and
+  much smaller than inlining the full feature.
+- When the plugin NAR is not installed, broker behavior is **byte-for-byte
+  identical** to unmodified Pulsar; the ~99% of users that never switch BK
+  clusters pay zero cost in core surface area.
+
+## Goal
+
+Provide a runtime mechanism by which a broker cluster can be re-pointed from
+one Bookie cluster (`oldCluster`) to a new Bookie cluster (`newCluster`)
+without service interruption, while preserving the ability to read historical
+data that physically resides on `oldCluster`.
+
+### In Scope
+
+- A plugin-provided `ManagedLedgerStorage` implementation
+  (`MultiClusterManagedLedgerStorage`) that holds a
+  `Map<clusterName, BookkeeperManagedLedgerStorageClass>` and exposes the
+  currently active one.
+- Per-ledger cluster attribution stored in the **existing**
+  `LedgerInfo.properties` and `ManagedCursorInfo.cursorProperties` slots under
+  the reserved key `_pulsar.bookieClusterName`.
+- Per-broker-cluster registry of multiple Bookie clusters, stored in
+  Broker-ZK under a plugin-owned path.
+- A plugin-owned `BookieClusterReadRouter` that selects a BK client at
+  ledger-handle open time based on the attribution property.
+- A plugin-owned `BookieClusterSwitchOrchestrator` that drives
+  `BUILD → PROMOTE → CLEANUP` for Schema and Cursor Ledgers (migration
+  preserves the original `ledgerId` via `BookKeeper.asyncCreateLedgerAdv`).
+- Plugin-provided CLI (`pulsar-admin bookie-clusters …`) via PIP-201
+  `CustomCommandFactory`.
+- Plugin-provided REST API under `/admin/v2/bookie-clusters/*` via
+  `AdditionalServlet`.
+- Three narrow core SPI hooks (see Section "Required Core SPI Additions").
+
+### Out of Scope
+
+- **Topic Data Ledger migration** — not migrated; ages out via TTL/Retention.
+- **Automatic failure detection and switch triggering** — operator-initiated
+  to avoid split-brain.
+- **Cross-cluster (geo-replicated) coordination** — each broker cluster
+  switches independently.
+- **BookKeeper-cluster-side internal migration tools** — delegated to BookKeeper.
+- **Changing `metadataStoreUrl` (Broker-ZK)** — only the Bookie cluster is
+  switched; Broker-ZK remains the same throughout.
+- **New proto fields in `MLDataFormats.proto`** — reuses existing extension
+  slots from PIP-404 and `cursorProperties`.
+
+## High-Level Architecture
+
+```
+┌────────────────────────────────────────────────────────────────────────┐
+│ Pulsar core (unchanged + 3 narrow SPI hooks)                           │
+│                                                                         │
+│   ManagedLedgerStorage SPI            ─── PIP-384                       │
+│   ManagedLedgerStorageClass SPI       ─── PIP-384                       │
+│   LedgerInfo.properties (KV)          ─── PIP-404                       │
+│   ManagedCursorInfo.cursorProperties  ─── pre-existing                  │
+│   AdditionalServlet SPI               ─── pre-existing                  │
+│   CustomCommandFactory SPI            ─── PIP-201                       │
+│                                                                         │
+│   + ManagedLedgerConfig.activeBookKeeperSupplier         ←── NEW (S1)   │
+│   + ManagedLedgerConfig.bookKeeperResolver               ←── NEW (S2)   │
+│   + SchemaStorageBookKeeperProvider SPI                  ←── NEW (S3)   │
+└────────────────────────────┬───────────────────────────────────────────┘
+                             │ SPI / NAR loading
+┌────────────────────────────▼───────────────────────────────────────────┐
+│ pulsar-bookie-cluster-switching (NAR, opt-in)                          │
+│                                                                         │
+│   MultiClusterManagedLedgerStorage   impl ManagedLedgerStorage         │
+│     ├─ Map<name, BookkeeperManagedLedgerStorageClass>                  │
+│     ├─ activeClusterName (volatile)                                     │
+│     └─ getDefaultStorageClass() → active                               │
+│                                                                         │
+│   RoutingManagedLedgerFactory        wraps ManagedLedgerFactoryImpl    │
+│     └─ Per-ledger BK resolution from LedgerInfo.properties             │
+│                                                                         │
+│   RoutingSchemaBookKeeperProvider    impl SchemaStorageBookKeeperProvider│
+│     └─ Per-position BK resolution from SchemaLocator extension         │
+│                                                                         │
+│   BookieClusterConfigManager / ConfigWatcher                            │
+│     └─ Owns ZK path /admin/bookie-clusters/* (plugin-owned)            │
+│                                                                         │
+│   BookieClusterSwitchOrchestrator    BUILD → PROMOTE → CLEANUP          │
+│                                                                         │
+│   REST  /admin/v2/bookie-clusters/*  via AdditionalServlet              │
+│   CLI   pulsar-admin bookie-clusters via CustomCommandFactory NAR       │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Key Principles
+
+| Principle | Description |
+|-----------|-------------|
+| **Attribution as Single Source of Truth** | Every ledger at creation is "stamped" with the active cluster name inside its existing metadata properties. Reads are routed by that stamp. No history table, no boundary heuristics, no side znode. |
+| **Reuse Extension Slots, Don't Add Proto Fields** | `LedgerInfo.properties` (PIP-404) and `ManagedCursorInfo.cursorProperties` carry the attribution. Only `SchemaStorageFormat.proto` needs one tiny extension (Section "Required Core SPI Additions" S3). |
+| **Topic Data: Don't Migrate** | Topic Data Ledgers have TTL/Retention; let them age out. Switching does not move terabytes of data. |
+| **Schema / Cursor: Migrate, but Preserve `ledgerId`** | We use `asyncCreateLedgerAdv(ledgerId, …)` to recreate the **same `ledgerId`** in the new cluster, copy entries with **identical `entryId`**, then CAS only the attribution property. Business keys (`cursorsLedgerId`, `position.ledgerId`, `position.entryId`) are byte-for-byte unchanged in Broker-ZK. |
+| **Plugin-First, Opt-In** | When the NAR is absent or `managedLedgerStorageClassName` ≠ the multi-cluster class, the broker is byte-for-byte unmodified Pulsar. |
+| **Operator-Driven, Single Active Cluster** | At any moment exactly one cluster is `status=ACTIVE` in Broker-ZK. Auto-failover is rejected. |
+| **Broker-ZK is the Switch Source-of-Truth** | All cluster registrations and the switch directive live in Broker-ZK under a plugin-owned path. No Global-ZK, no external coordinator. |
+
+## Required Core SPI Additions
+
+These are the **only** changes to Pulsar core required by this PIP. Each is a
+narrow, generally-useful hook that any custom `ManagedLedgerStorage`
+implementation could leverage — not specific to BK switching.
+
+### S1. `ManagedLedgerConfig.activeBookKeeperSupplier`
+
+**Problem.** Today `ManagedLedgerImpl` captures the `BookKeeper` client once
+in its constructor (via `ManagedLedgerFactoryImpl.bookkeeperFactory`). When a
+custom storage class needs to hot-swap the underlying BK client (e.g. cluster
+switch), there is no clean injection point on the write path.
+
+**Hook.** Add an optional supplier:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Supplier<BookKeeper> activeBookKeeperSupplier;   // nullable
+
+public Supplier<BookKeeper> getActiveBookKeeperSupplier() { … }
+public ManagedLedgerConfig setActiveBookKeeperSupplier(Supplier<BookKeeper> s) { … }
+```
+
+`ManagedLedgerImpl.createLedgerAfterClosed`, `rollCurrentLedgerIfFull`, and
+`ManagedCursorImpl.doCreateNewMetadataLedger` resolve the BK client as:
+
+```java
+BookKeeper bk = config.getActiveBookKeeperSupplier() != null
+        ? config.getActiveBookKeeperSupplier().get()
+        : this.bookKeeper;     // backward-compatible default
+```
+
+**Scope.** Pure additive. When the supplier is unset (default), behavior is
+identical. ~30 lines of change in `managed-ledger/`.
+
+### S2. `ManagedLedgerConfig.bookKeeperResolver`
+
+**Problem.** Reads (and deletes) need a *per-ledger* resolution path. Today
+`ManagedLedgerImpl` uses the constructor-captured `bookKeeper` for every
+`asyncOpenLedger`/`asyncDeleteLedger`. After a switch, ledgers that physically
+live on the old cluster must be opened with the old client.
+
+**Hook.** Add an optional resolver function keyed on `LedgerInfo`:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Function<LedgerInfo, BookKeeper> bookKeeperResolver;   // nullable
+
+// And the analogous function for cursor metadata ledgers:
+private Function<ManagedCursorInfo, BookKeeper> cursorBookKeeperResolver;
+```
+
+All call sites that today reference `this.bookKeeper` to open or delete a
+specific ledger (the comprehensive list is in
+`ManagedLedgerImpl.getLedgerHandle`, `internalAsyncOpenCursor`,
+`asyncDeleteLedger`, `asyncDeleteFromBookKeeper`, …) become:
+
+```java
+BookKeeper bk = resolveBookKeeperForLedger(ledgerInfo);
+```
+
+Where:
+
+```java
+private BookKeeper resolveBookKeeperForLedger(LedgerInfo li) {
+    var resolver = config.getBookKeeperResolver();
+    return resolver != null ? resolver.apply(li) : this.bookKeeper;
+}
+```
+
+**Scope.** Pure additive; when resolvers are unset, behavior is identical.
+~80 lines of change in `managed-ledger/`, mostly mechanical substitutions.
+
+### S3. `SchemaStorageBookKeeperProvider` SPI
+
+**Problem.** `BookkeeperSchemaStorage` captures `this.bookKeeper` at
+`start()` time (line ~101). It directly opens schema ledgers via that single
+client. To support per-ledger routing we need to inject a resolver and to
+attribute each schema ledger to a cluster.
+
+**Hook (two parts).**
+
+1. **Add a tiny extension slot to `SchemaStorageFormat.proto`** — this is the
+   *only* proto change in this PIP:
+
+   ```protobuf
+   message PositionInfo {
+       required int64 ledgerId = 1;
+       required int64 entryId = 2;
+       repeated KeyValue properties = 3;   // ★ NEW — generic extension slot,
+                                           // mirrors PIP-404 LedgerInfo.properties
+   }
+
+   message KeyValue {
+       required string key = 1;
+       required string value = 2;
+   }
+   ```
+
+   This is the minimal possible change: a generic `properties` slot symmetric
+   to PIP-404. It does **not** mention bookie clusters; any future plugin can
+   use it.
+
+2. **Add a provider SPI for the schema storage BK client:**
+
+   ```java
+   // pulsar-broker/.../service/schema/SchemaStorageBookKeeperProvider.java
+   public interface SchemaStorageBookKeeperProvider {
+       /** Returns the BK client to use for the given schema position. */
+       BookKeeper resolve(PositionInfo position);
+       /** Returns the BK client to use for *creating* a new schema ledger. */
+       BookKeeper active();
+   }
+   ```
+
+   `BookkeeperSchemaStorage` resolves its provider as:
+
+   ```java
+   var provider = pulsar.getSchemaStorageBookKeeperProvider();   // nullable
+   BookKeeper bk = provider != null ? provider.resolve(position) : this.bookKeeper;
+   ```
+
+   The default (provider == null) preserves today's behavior exactly.
+
+**Scope.** One proto field (additive, `repeated` is forward/backward
+compatible in proto2) + one SPI interface + ~50 lines in
+`BookkeeperSchemaStorage`. No new REST endpoint, no CLI change.
+
+### Summary: Core Diff Footprint
+
+| Component | LOC | Risk |
+|-----------|-----|------|
+| `ManagedLedgerConfig` (S1, S2 setters/getters) | ~40 | Trivial additive |
+| `ManagedLedgerImpl` / `ManagedCursorImpl` resolver wiring | ~80 | Mechanical, well-scoped |
+| `SchemaStorageFormat.proto` `PositionInfo.properties` (S3) | 1 field | Additive, proto2-safe |
+| `SchemaStorageBookKeeperProvider` SPI + `BookkeeperSchemaStorage` wiring | ~50 | Additive |
+| **Total core diff** | **~170 LOC, 1 new optional proto field** | **Zero behavioral change when SPI unused** |
+
+Everything else lives in the plugin NAR.
+
+## Plugin Module: `pulsar-bookie-cluster-switching`
+
+A new top-level Maven module that builds a NAR (`pulsar-bookie-cluster-switching-<version>.nar`).
+
+### Module Layout
+
+```
+pulsar-bookie-cluster-switching/
+├─ src/main/java/.../
+│   ├─ storage/
+│   │   ├─ MultiClusterManagedLedgerStorage.java    (implements ManagedLedgerStorage)
+│   │   ├─ MultiClusterStorageClass.java            (implements BookkeeperManagedLedgerStorageClass)
+│   │   └─ RoutingManagedLedgerFactory.java
+│   ├─ schema/
+│   │   └─ RoutingSchemaBookKeeperProvider.java     (implements SchemaStorageBookKeeperProvider)
+│   ├─ registry/
+│   │   ├─ BookieClusterConfigManager.java
+│   │   └─ BookieClusterConfigWatcher.java
+│   ├─ orchestrator/
+│   │   ├─ BookieClusterSwitchOrchestrator.java
+│   │   ├─ LedgerCopyUtil.java
+│   │   └─ IdgenPrecheckService.java
+│   ├─ rest/
+│   │   └─ BookieClusterAdminServlet.java           (implements AdditionalServlet)
+│   └─ cli/
+│       └─ BookieClusterCommandFactory.java         (implements CustomCommandFactory)
+└─ src/main/resources/META-INF/services/
+    ├─ org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet
+    └─ org.apache.pulsar.admin.cli.extensions.CustomCommandFactory
+```
+
+### Wiring on Broker Start
+
+The plugin is activated by setting in `broker.conf`:
+
+```properties
+managedLedgerStorageClassName=org.apache.pulsar.ext.bookieswitch.MultiClusterManagedLedgerStorage
+additionalServlets=bookie-clusters-admin
+additionalServletDirectory=./plugins
+```
+
+Then `ManagedLedgerStorage.create()` (PIP-384 entry point) reflectively loads
+`MultiClusterManagedLedgerStorage`. Inside its `initialize()`:
+
+1. Reads its own configuration namespace (prefix `bookieClusterSwitch.*`).
+2. Reads `/admin/bookie-clusters/*` from Broker-ZK to discover registered
+   clusters and resolve the initial `activeClusterName` (tier-1 →
+   Broker-ZK `status=ACTIVE` cluster; tier-2 → optional
+   `bookieClusterSwitch.currentClusterName` hint; tier-3 → derive from
+   `bookkeeperMetadataServiceUri` and auto-register as `ACTIVE`).
+3. Builds one `BookkeeperManagedLedgerStorageClass` per registered cluster
+   (each owns its own `BookKeeper` client and `StatsProvider`).
+4. Constructs a `RoutingManagedLedgerFactory` that decorates a base
+   `ManagedLedgerFactoryImpl` and injects the per-ledger BK resolver via the
+   S1/S2 hooks above.
+5. Starts `BookieClusterConfigWatcher` to receive switch directives.
+6. (Origin-broker-only after a switch) starts the orchestrator.
+
+The plugin's `AdditionalServlet` instance is independently discovered by
+`AdditionalServlets.load(…)` in `WebService`; the plugin's
+`CustomCommandFactory` is independently discovered by `pulsar-admin`'s NAR
+class loader.
+
+## Detailed Design
+
+### Per-Ledger Attribution Encoding
+
+| Ledger type | proto container | Encoding | Key |
+|-------------|-----------------|----------|-----|
+| Topic Data | `ManagedLedgerInfo.LedgerInfo.properties` (PIP-404, tag 6) | `KeyValue` | `_pulsar.bookieClusterName` |
+| Cursor | `ManagedCursorInfo.cursorProperties` (existing, tag 8) | `StringProperty` | `_pulsar.bookieClusterName` |
+| Schema | `SchemaStorageFormat.PositionInfo.properties` (NEW, tag 3, S3) | `KeyValue` | `_pulsar.bookieClusterName` |
+
+The reserved key prefix `_pulsar.` is **owned by Pulsar core** to avoid
+collisions with plugin- or user-defined properties. Plugin code uses a
+constant:
+
+```java
+public static final String ATTR_KEY = "_pulsar.bookieClusterName";
+```
+
+**Compatibility.**
+
+- **proto2 forward-compat:** `repeated KeyValue properties` already exists in
+  `LedgerInfo` (PIP-404). For `PositionInfo` it's added with a new tag (3),
+  matching the symmetric pattern.
+- **Old broker reading new metadata:** ignores `properties` it doesn't
+  understand; preserves them via `UnknownFieldSet` on rewrite (already the
+  case for PIP-404).
+- **New broker reading old metadata (no attribution):** the plugin's resolver
+  falls back to the active cluster (which equals the original cluster for
+  pre-switch ledgers; correct).
+
+### Write-Path Stamping
+
+`RoutingManagedLedgerFactory` wraps the base factory. When `ManagedLedger`
+asks for a new ledger:
+
+```java
+// Pseudocode of the stamping path
+String activeCluster = multiClusterStorage.getActiveClusterName();
+BookKeeper bk = multiClusterStorage.getBookKeeper(activeCluster);    // via S1
+LedgerHandle lh = bk.asyncCreateLedger(...);
+
+// In the same metadata CAS that records the new LedgerInfo:
+LedgerInfo info = baseInfo.toBuilder()
+    .addProperties(KeyValue.newBuilder()
+        .setKey(ATTR_KEY)
+        .setValue(activeCluster)
+        .build())
+    .build();
+```
+
+The same pattern applies to:
+
+- `ManagedCursorImpl.doCreateNewMetadataLedger` → stamps
+  `cursorProperties[_pulsar.bookieClusterName] = activeCluster`.
+- `BookkeeperSchemaStorage.createLedger` (via S3 provider) → stamps
+  `PositionInfo.properties[_pulsar.bookieClusterName] = activeCluster`.
+
+The stamp is written in the **same Broker-ZK CAS** as the business field
+(`ledgerId`, `cursorsLedgerId`, `position`), guaranteeing atomicity
+(invariant **I1**).
+
+### Read-Path Routing
+
+```java
+// RoutingManagedLedgerFactory -- injected as ManagedLedgerConfig.bookKeeperResolver (S2)
+BookKeeper resolve(LedgerInfo li) {
+    String cluster = li.getPropertiesList().stream()
+        .filter(kv -> ATTR_KEY.equals(kv.getKey()))
+        .map(KeyValue::getValue)
+        .findFirst()
+        .orElseGet(multiClusterStorage::getActiveClusterName);   // legacy fallback
+    return multiClusterStorage.getBookKeeper(cluster);
+}
+```
+
+Identical pattern for cursors (consulting `cursorProperties`) and schema
+positions (consulting `PositionInfo.properties`).
+
+**Invariant I4** (entry-level consistency within a single read): the resolver
+is consulted exactly once at `LedgerHandle` open time; the resulting
+`LedgerHandle` is bound to one BK client for the entire read of that ledger.
+
+### Switch Phases (Operator-Driven State Machine)
+
+```
+NOT_REGISTERED
+    ↓
+[Operator: pulsar-admin bookie-clusters register --name newCluster --uri … --status STANDBY]
+    ↓
+REGISTERED (STANDBY)
+    ↓
+[Operator: pulsar-admin bookie-clusters switch --target newCluster]
+    ↓ (inline precheck: idgen-long advanced)
+SWITCH_TRIGGERED ← Coordinator broker writes /admin/bookie-clusters/switch-target
+    ↓
+LIVE_DUAL_READ  ← All brokers' watchers fire; new writes stamped newCluster;
+    ↓             reads of pre-stamped ledgers routed by attribution
+BUILD_PROMOTE_CLEANUP
+    ├── BUILD     ← Coordinator copies Schema/Cursor ledger bytes to newCluster
+    │             with the SAME ledgerId; entries copied with the SAME entryId
+    ├── PROMOTE   ← Schema: coordinator CAS-rewrites SchemaLocator
+    │             (changes only PositionInfo.properties[ATTR_KEY])
+    │             Cursor: coordinator forwards POST /internal/promote-cursor
+    │             to topic owner; owner CAS in ManagedCursorImpl lock
+    └── CLEANUP   ← After rollback window, delete old-cluster copies; promote
+                  Broker-ZK status: oldCluster→DEPRECATED, newCluster→ACTIVE
+    ↓
+DONE / DONE_WITH_FAILURES
+
+(Operator-triggered, within window):
+ROLLBACK ← Reverse-CAS attribution back to oldCluster; new-cluster
+          copies scheduled for delayed deletion.
+```
+
+All states above live **inside the plugin**. The orchestrator persists its
+progress under `/admin/bookie-clusters/orchestrator/<switchId>/...`.
+
+### Phase 1: Registration
+
+```bash
+pulsar-admin bookie-clusters register \
+    --name new-cluster \
+    --metadata-service-uri zk+null://new-zk:2181/ledgers-v2 \
+    --status STANDBY
+```
+
+`BookieClusterConfigManager` (plugin) validates:
+
+- `metadata-service-uri` does not collide with any already-registered cluster.
+- For co-located deployments, the chroot must differ from existing clusters.
+
+### Phase 2: Precheck (Invariant I11)
+
+```bash
+pulsar-admin bookie-clusters precheck --name new-cluster
+```
+
+`IdgenPrecheckService` (plugin) verifies the new cluster's `/ledgers/idgen-long`
+is advanced beyond `max(ledgerId)` of the source cluster. Otherwise
+`asyncCreateLedgerAdv(ledgerId, …)` during BUILD would collide with
+newly-allocated ledgerIds. `POST /switch` invokes precheck inline; failure
+returns HTTP 409 (bypassable via plugin config in lab/staging only).
+
+### Phase 3: Switch Trigger
+
+```bash
+pulsar-admin bookie-clusters switch --target new-cluster
+```
+
+The receiving broker (origin) writes:
+
+```
+/admin/bookie-clusters/switch-target
+  body: { "target": "new-cluster", "origin": "https://broker-1:8080",
+          "epoch": 7, "ts": 1716256800000 }
+```
+
+All brokers' `BookieClusterConfigWatcher` fire and execute, in order:
+
+1. `MultiClusterManagedLedgerStorage.switchActiveCluster(newCluster)` —
+   hot-swap `activeClusterName`; new BK client lazily created if not present.
+2. First switch: enable dual-read in the read router; subsequent switches:
+   refresh epoch.
+3. Emit metrics.
+4. **Origin broker only:** start the orchestrator (`BUILD → PROMOTE → CLEANUP`).
+   Non-origin brokers do not start an orchestrator. If the directive body
+   lacks `origin`, brokers fall back to leader election among themselves
+   (using the existing `LeaderElectionService`) to pick exactly one
+   orchestrator.
+
+### Phase 4: BUILD (Schema + Cursor Ledger Bytes)
+
+The orchestrator runs **two parallel stages** with bounded concurrency.
+
+#### Schema Ledger BUILD (centralized)
+
+```
+for each schemaId in /schemas/* (parallel, bounded):
+  locator = brokerZk.get("/schemas/<schemaId>")
+  for each indexEntry in locator.indexList:
+    ledgerId = indexEntry.position.ledgerId
+    cluster  = readClusterAttr(indexEntry.position) ?? activeClusterAtFirstWrite
+    if cluster == oldCluster:
+      LedgerCopyUtil.copyLedgerPreservingIds(
+          ledgerId, oldBk, newBk,
+          customMetadata,           // identical to source ledger
+          ensembleSize, qw, qa)
+        // Internally:
+        //   newBk.asyncCreateLedgerAdv(ledgerId, …)        ← SAME ledgerId
+        //   for entryId in [0, lastAddConfirmed]:
+        //     newLh.addEntry(entryId, srcEntry)            ← SAME entryId
+        //   newLh.close()
+```
+
+#### Cursor Ledger BUILD (centralized)
+
+Identical pattern over `/managed-ledgers/<topic>/<cursorName>` znodes, copying

Review Comment:
   Could we discuss whether cursor metadata needs to be migrated by copying the existing cursor ledger with the same ledgerId?
   
   If the topic owner can force a cursor metadata checkpoint/rollover after the active BK cluster changes, it could create a new cursor metadata ledger on the target BK, persist the current mark-delete position / deleted ranges / cursor properties, and CAS-update `ManagedCursorInfo.cursorsLedgerId`.
   
   Would that be sufficient, or is there a reason the old cursor ledger needs to be byte-copied with `asyncCreateLedgerAdv`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-470: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["dao-jun (via GitHub)" <gi...@apache.org>]

dao-jun commented on code in PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#discussion_r3241892479


##########
pip/pip-470.md:
##########
@@ -0,0 +1,587 @@
+# PIP-470: Dynamic Bookie Cluster Switching for Brokers
+
+## Motivation
+
+Apache Pulsar's broker layer is bound to a fixed BookKeeper (Bookie) cluster at startup time through the `bookkeeperMetadataServiceUri` configuration. This static binding becomes a bottleneck in several real-world operational scenarios:
+
+1. **Slow Failure Recovery**: When a Bookie cluster experiences capacity exhaustion, sustained latency degradation, or non-recoverable failures, the only mitigation today is to wait for the Bookie cluster to recover. In elastic-storage deployments where a single Bookie cluster backs 100+ broker clusters, the blast radius is enormous.
+
+2. **No Capacity Escape Hatch**: When a Bookie cluster approaches its capacity ceiling, brokers cannot "spill over" to a fresh Bookie cluster. Operators must scale the existing cluster in place, which is slow and risky.
+
+3. **No Runtime Re-targeting**: There is no supported runtime mechanism to migrate brokers from one Bookie cluster to another without restart and full data migration.
+
+4. **Data Plane Coupling**: Bookie cluster switching deals with the data plane: terabytes of ledger bytes, ongoing publishes, ongoing acknowledgments, and live cursors. A copy-everything-then-cutover approach is infeasible.
+
+This PIP proposes a **safe, ledger-attribution-driven framework** that allows operators to switch a broker cluster's underlying Bookie cluster at runtime, with **zero data-plane downtime** and **per-ledger routing** that preserves the readability of historical data on the old cluster.
+
+The framework enables:
+
+- **Zero-downtime switching** — publish and consume continue without interruption
+- **Per-ledger routing** — each ledger carries its own cluster attribution; reads are routed to the cluster that physically holds the bytes
+- **No bulk data migration** — Topic Data Ledgers are NOT migrated; they age out naturally via TTL/Retention
+- **Targeted migration of small, long-lived metadata ledgers** — only Schema Ledgers and Cursor Ledgers are migrated, while preserving their original `ledgerId`
+- **Automatic rollback** within a configurable window
+- **Operator-driven, not auto-failover** — avoids split-brain
+
+## Goal
+
+Provide a runtime mechanism by which a broker cluster can be re-pointed from one Bookie cluster (`oldCluster`) to a new Bookie cluster (`newCluster`) without service interruption, while preserving the ability to read historical data that physically resides on `oldCluster`.
+
+### In Scope
+
+- Per-broker-cluster configuration of multiple registered Bookie clusters via Broker-ZK metadata
+- A `DynamicBookKeeperClientFactory` that holds multiple BK clients simultaneously
+- Per-ledger cluster attribution stored in proto fields (`LedgerInfo.bookieClusterName`, `ManagedCursorInfo.bookieClusterName`, `PositionInfo.bookieClusterName`)
+- Read-path routing keyed by attribution field (`BookieClusterReadRouter`)
+- Asynchronous post-switch migration of Schema and Cursor Ledgers, **preserving the original `ledgerId`** via `BookKeeper.asyncCreateLedgerAdv(ledgerId, ...)`
+- CLI (`pulsar-admin bookie-clusters`) + REST API + Java Admin SDK (three-layer consistency)
+- Automatic rollback within a configurable window
+- Support for both **co-located deployments** (Broker-ZK and Bookie-ZK share the same physical ZK ensemble with different chroots) and **split deployments**
+
+### Out of Scope
+
+- **Topic Data Ledger migration**: Topic data is NOT migrated; old Topic Data Ledgers age out via the topic's TTL/Retention policy. Old data is read from the old cluster via attribution-based routing for as long as the old cluster is reachable.
+- **Automatic failure detection and switch triggering**: Switching is operator-initiated to avoid split-brain.
+- **Cross-cluster (geo-replicated) coordination**: Each broker cluster switches its own Bookie cluster independently.
+- **BookKeeper-cluster-side internal migration tools**: This PIP only orchestrates from the broker side; recovering data within a Bookie cluster (e.g., underreplication) is delegated to BookKeeper itself.
+- **Changing `metadataStoreUrl` (Broker-ZK)**: Only the Bookie cluster (`bookkeeperMetadataServiceUri`) is switched; Broker-ZK remains the same throughout.
+
+## High Level Design
+
+The framework introduces three new long-lived broker components:
+
+1. **`DynamicBookKeeperClientFactory`** — Maintains a `Map<clusterName, BookKeeper>` and exposes the currently active client. Replaces the singleton `BookKeeper` previously held by `ManagedLedgerClientFactory`.
+2. **`BookieClusterReadRouter`** — On every read, resolves the target cluster from the per-ledger attribution field and returns the matching BK client.
+3. **`BookieClusterSwitchOrchestrator`** — Drives the post-switch state machine (`BUILD → PROMOTE → CLEANUP`) for Schema and Cursor Ledgers.
+
+### Key Principles
+
+| Principle | Description |
+|-----------|-------------|
+| **BookieClusterName as Single Source of Truth** | At ledger creation time, the active cluster name is "stamped" into the ledger's metadata. Reads are routed by that stamp. No history table, no boundary heuristics. |
+| **Attribution Field, Not Side ZNode** | The attribution lives directly inside `MLDataFormats.proto`'s `LedgerInfo` / `ManagedCursorInfo` and `SchemaStorageFormat.proto`'s `PositionInfo`. No "/admin/bookie-clusters/schema-ledger-clusters" side path is introduced. This eliminates dual-source-of-truth and cache-coherence risks. |
+| **Topic Data: Don't Migrate** | Topic Data Ledgers have TTL/Retention; let them age out. Switching does not move terabytes of data. |
+| **Schema / Cursor: Migrate, but Preserve `ledgerId`** | Schema/Cursor are tied to topic/subscription lifecycles and cannot be left behind. We use `asyncCreateLedgerAdv(ledgerId, ...)` to recreate the **same `ledgerId`** in the new cluster, copy entries with **identical `entryId`**, then CAS only the `bookieClusterName` field. **`cursorsLedgerId`, `position.ledgerId`, `position.entryId` are byte-for-byte unchanged** in Broker-ZK. |
+| **Operator-Driven, Single Active Cluster** | Switching is triggered by `pulsar-admin bookie-clusters switch`; auto-failover is rejected. At any moment exactly one cluster is `status=ACTIVE` in Broker-ZK. |
+| **Broker-ZK is the Switch Source-of-Truth** | All cluster registrations and the switch directive live in Broker-ZK. No Global-ZK, no external coordinator. |
+| **Minimum proto Change** | Only `optional` fields are added (`bookieClusterName`, `properties`). proto2 forward/backward compatibility holds. |
+
+## Detailed Design
+
+### Switch Phases
+
+```
+NOT_REGISTERED
+    ↓
+[Operator: pulsar-admin bookie-clusters register --name newCluster --uri ... --status STANDBY]
+    ↓
+REGISTERED (STANDBY)
+    ↓
+[Operator: pulsar-admin bookie-clusters switch --target newCluster]
+    ↓ (POST /switch internally invokes /precheck for idgen-long, see I11)
+SWITCH_TRIGGERED ← Coordinator broker writes /admin/bookie-clusters/switch-target
+    ↓
+LIVE_DUAL_READ ← All brokers' watchers fire; new writes go to newCluster;
+    ↓             reads still go to oldCluster for ledgers attributed to it
+BUILD_PROMOTE_CLEANUP
+    ├── BUILD     ← Coordinator centrally copies Schema/Cursor ledger bytes to newCluster
+    │             with the SAME ledgerId; entries copied with the SAME entryId
+    ├── PROMOTE   ← Schema: coordinator CAS-rewrites SchemaLocator (changes only
+    │             position.bookieClusterName)
+    │             Cursor: coordinator forwards POST /internal/promote-cursor
+    │             to topic owner broker; owner does CAS in ManagedCursorImpl lock
+    └── CLEANUP   ← After rollback window, delete old-cluster copies; promote
+                  Broker-ZK status: oldCluster→DEPRECATED, newCluster→ACTIVE
+    ↓
+DONE / DONE_WITH_FAILURES
+
+(Operator-triggered, within rollback window):
+ROLLBACK ← CAS attribution fields back to oldCluster; new-cluster copies are
+          deleted asynchronously
+```
+
+### Phase 1: Cluster Registration
+
+```bash
+pulsar-admin bookie-clusters register \
+    --name new-cluster \
+    --metadata-service-uri zk+null://new-zk:2181/ledgers-v2 \
+    --status STANDBY
+```
+
+The `BookieClusterConfigManager` validates that:
+- The `metadata-service-uri` does not collide with any already-registered cluster (prevents accidental aliasing in the BK client map).
+- For co-located deployments, the chroot must differ from existing clusters.
+
+### Phase 2: Pre-Switch Validation (`precheck`)
+
+```bash
+pulsar-admin bookie-clusters precheck --name new-cluster
+```
+
+`IdgenPrecheckService` verifies invariant **I11**: the new cluster's `/ledgers/idgen-long` must be advanced beyond `max(ledgerId)` of the source cluster. Otherwise, `asyncCreateLedgerAdv(ledgerId, ...)` during BUILD will collide with newly-allocated ledgerIds. The precheck returns a structured result:
+
+```json
+{
+  "ready": true,
+  "idgenAdvancedBeyondSource": true,
+  "sourceClusterName": "old-cluster",
+  "targetClusterName": "new-cluster",
+  "sourceMaxHighOrderBit": 12345,
+  "targetMaxHighOrderBit": 1012345
+}
+```
+
+`POST /switch` invokes `precheck` inline; failure returns HTTP 409 (bypassable via `bookieClusterSwitchSkipIdgenPrecheck=true` in lab/staging only).
+
+### Phase 3: Switch Trigger
+
+```bash
+pulsar-admin bookie-clusters switch --target new-cluster
+```
+
+The receiving broker (origin) writes the `switch-target` znode in Broker-ZK with CAS:
+
+```
+/admin/bookie-clusters/switch-target
+  body: "new-cluster\norigin=https://broker-1:8080"
+```
+
+All brokers' `BookieClusterConfigWatcher` fire, executing the callback chain:
+
+1. `DynamicBookKeeperClientFactory.switchToCluster(newCluster)` — hot-swap `activeClusterName`; new BK client lazily created if not yet present
+2. First switch: `BookieClusterReadRouter.enableDualRead()` (records `initialClusterName`, `switchLedgerIdBoundary`, `switchTimestampMillis`)
+   Subsequent switches: `BookieClusterReadRouter.recordSwitchEpoch()` (only refreshes `switchTimestampMillis`)
+3. `BookieClusterSwitchMetrics.recordSwitchSuccess()`
+4. **Origin broker only**: `BookieClusterSwitchOrchestrator.runStateMachine(BUILD → PROMOTE → CLEANUP)`
+
+Non-origin brokers do not start the orchestrator. If the `switch-target` znode body lacks the `origin=` line (e.g., legacy or operator-edited), brokers fall back to leader election to pick exactly one orchestrator.
+
+### Phase 4: BUILD (Schema + Cursor Ledger Bytes Copy)
+
+The orchestrator runs **two stages in parallel** with bounded concurrency (default 16):
+
+#### Schema Ledger BUILD (centralized)
+
+```
+for each schemaId in /schemas/* (parallel, bounded):
+  locator = brokerZk.get("/schemas/<schemaId>")
+  for each indexEntry in locator.indexList:
+    ledgerId = indexEntry.position.ledgerId
+    cluster  = indexEntry.position.bookieClusterName ?? activeCluster
+    if cluster == oldCluster:
+      LedgerCopyUtil.copyLedgerPreservingIds(
+          ledgerId, oldBk, newBk,
+          customMetadata,  // identical to source ledger
+          ensembleSize, qw, qa)
+        // Internally:
+        //   newBk.asyncCreateLedgerAdv(ledgerId, ...)        ← SAME ledgerId
+        //   for entryId in [0, lastAddConfirmed]:
+        //     newLh.addEntry(entryId, srcEntry)              ← SAME entryId
+        //   newLh.close()
+```
+
+#### Cursor Ledger BUILD (centralized)
+
+Identical pattern, scanning `/managed-ledgers/<topic>/<cursorName>` znodes, copying each `cursorsLedgerId`.
+
+**Why BUILD is safe to centralize**:
+1. Reads of the old ledger are read-only — no contention with owner brokers' write paths.
+2. Writes to the new cluster are to a **dormant copy** — no online reader sees it yet (attribution is still `oldCluster`).
+3. The orchestrator uses dedicated BK clients with isolated thread pools.
+
+#### Idempotency on `LedgerExistException`
+
+If `asyncCreateLedgerAdv(ledgerId, ...)` returns `LedgerExistException` (BUILD retry / partial prior run), `LedgerCopyUtil.verifyDstMatchesSrc` opens both ledgers and compares `getLastAddConfirmed()` and `getLength()`. If both match, the existing ledger is treated as a successful idempotent copy (`migration_ledger_id_conflict_total` is incremented for visibility). Mismatch → record as `failedLedgers`; attribution remains on the old cluster.
+
+### Phase 5: PROMOTE (CAS Attribution Field)
+
+#### Schema PROMOTE (centralized)
+
+```
+for each schemaId built successfully:
+  retry up to MAX_PROMOTE_RETRIES (5):
+    (locator, version) = brokerZk.getWithVersion("/schemas/<schemaId>")
+    updated = locator.toBuilder()
+        .clearIndex()
+        .addAllIndex(locator.indexList.map { ie ->
+            if (ie.position.bookieClusterName == oldCluster):
+              ie.toBuilder().setPosition(
+                  ie.position.toBuilder()
+                      .setBookieClusterName(newCluster)  // ★ ONLY field changed
+                      .build()
+              ).build()
+            else: ie  // already newCluster (concurrent putSchema) — leave alone
+        })
+        .build()
+    try brokerZk.put("/schemas/<schemaId>", updated.toByteArray(), version)
+    catch BadVersionException: continue  // re-read and retry
+```
+
+**Crucially**, `position.ledgerId`, `position.entryId`, `info.version`, and `info.hash` are **byte-for-byte unchanged** (invariant **I10**). Only `bookieClusterName` is rewritten.
+
+#### Cursor PROMOTE (forwarded to topic owner)
+
+Cursor is more delicate because `ManagedCursorImpl` is a **stateful object owned by exactly one broker** (the topic owner) and holds an in-process lock that serializes markDelete flushes. Letting the orchestrator broker write `ManagedCursorInfo` directly would race with the owner's pending markDelete CAS.
+
+```
+[orchestrator broker]
+  for each (topic, cursorName, cursorsLedgerId) built successfully:
+    ownerUrl = namespaceService.findOwnerHttpUrl(topic)
+    POST {ownerUrl}/admin/v2/bookie-clusters/internal/promote-cursor
+        body: {topic, cursorName, targetClusterName=newCluster,
+               expectedCursorsLedgerId=cursorsLedgerId}
+
+[owner broker]
+  cursor = brokerService.getTopic(topic).getCursor(cursorName)
+  synchronized (cursor):
+    if cursor.cursorsLedgerId != expectedCursorsLedgerId:
+      return 409  // cursor rolled over since BUILD; coordinator records as
+                  // promotedViaRolloverFallback (the new cursorsLedgerId
+                  // was already stamped into newCluster at rollover time
+                  // by the writer-side attribution stamping)
+    metaStore.asyncUpdateCursorInfo(
+        cursor.path,
+        ManagedCursorInfo.newBuilder(cursor.managedCursorInfo)
+            .setBookieClusterName(newCluster)  // ★ ONLY field changed
+            .build(),
+        expectedZkVersion)
+    // BadVersion → re-read; bookieClusterName is orthogonal to
+    //   markDeleteLedgerId/markDeleteEntryId, so retry converges.
+```
+
+**Invariant I9**: PROMOTE must change only `bookieClusterName`; `cursorsLedgerId`, `markDeleteLedgerId`, `markDeleteEntryId`, individual deleted ranges, batched deletion indices, and cursor properties are preserved.
+
+#### Concurrent putSchema and PROMOTE
+
+For Schema, concurrent `putSchema` writes are naturally serialized via the `/schemas/<schemaId>` znode CAS:
+- If `putSchema` lands first, PROMOTE re-reads and sees the newly appended `IndexEntry` (already stamped `newCluster` because writes go to the active BK client) — leaves it alone, rewrites the rest.
+- If PROMOTE lands first, `putSchema` re-reads and appends a new `IndexEntry` with `newCluster` stamp.
+
+Either way, the converged state is "all entries stamped `newCluster`".
+
+### Phase 6: CLEANUP
+
+After the rollback window (`bookieClusterDualReadTimeoutDays`, default 15) elapses, or upon explicit `pulsar-admin bookie-clusters cleanup`:
+
+1. Delete old-cluster ledgers via `oldBk.asyncDeleteLedger(ledgerId)`.
+2. CAS-flip Broker-ZK status: `oldCluster.status = DEPRECATED`, `newCluster.status = ACTIVE` (invariant **I12**, two sequential CAS — `MetadataStore` API does not expose multi-op).
+3. Optionally close the old BK client via `pulsar-admin bookie-clusters close-client --name oldCluster`.
+
+`POST /admin/v2/bookie-clusters/cleanup` accepts:
+
+```json
+{
+  "dryRun": true,           // default; preview only
+  "srcClusterFilter": "...", // optional; restrict to one source cluster
+  "maxDeletions": 0,         // 0 = unlimited
+  "force": false             // bypass dual-read-window guard (lab only)
+}
+```
+
+### Phase 7: ROLLBACK (within window)
+
+If issues are detected, the operator triggers rollback. The reverse-CAS flips `bookieClusterName` back to `oldCluster` for affected Schema/Cursor; new-cluster ledgers are scheduled for delayed deletion. Because **`ledgerId` and `entryId` were never touched** in Broker-ZK, rollback is a constant-time metadata flip — no data movement.
+
+### Failure Handling
+
+| Stage | Failure | Behavior |
+|-------|---------|----------|
+| precheck | idgen-long not advanced | HTTP 409; switch refused |
+| Switch trigger | New BK client init fails | `activeClusterName` stays on old; switch rejected |
+| BUILD | Source ledger unreadable | Record in `failedLedgers`; attribution stays on old cluster (still readable) |
+| BUILD | `LedgerExistException` + LAC mismatch | Record in `failedLedgers`; do not PROMOTE; operator can retry switch |
+| PROMOTE | CAS BadVersion (5 retries exhausted) | Log warn; not counted in `failedLedgers`; next `switch --target <same>` retries (orchestrator is idempotent) |
+| PROMOTE | Cursor owner returns 409 (rollover) | Recorded as `promotedViaRolloverFallback`; old `cursorsLedgerId` ages out naturally |
+| CLEANUP | `promoteActiveCluster` CAS fails | Orchestrator finishes in `DONE` / `DONE_WITH_FAILURES`; in-memory `activeClusterName` is correct, but Broker-ZK ACTIVE may diverge → operator runs `register --status ACTIVE` to reconcile |
+
+## Implementation Details
+
+### Data Structures
+
+#### Cluster Registry (Broker-ZK)
+
+```
+/admin/bookie-clusters/                ← bookieClusterConfigPath (configurable)
+    ├── old-cluster                    ← znode body = JSON
+    │     {
+    │       "name": "old-cluster",
+    │       "metadataServiceUri": "zk://old-zk:2181/ledgers",
+    │       "status": "ACTIVE"          // exactly ONE cluster ACTIVE at a time
+    │     }
+    └── new-cluster
+          {
+            "name": "new-cluster",
+            "metadataServiceUri": "zk+null://new-zk:2181/ledgers-v2",
+            "status": "STANDBY"
+          }
+
+/admin/bookie-clusters/switch-target   ← written on switch
+    body: "new-cluster\norigin=https://broker-1:8080"
+```
+
+#### proto Changes
+
+**`managed-ledger/src/main/proto/MLDataFormats.proto`**:
+
+```protobuf
+message ManagedLedgerInfo {
+  message LedgerInfo {
+    required int64 ledgerId = 1;
+    optional int64 entries = 2;
+    optional int64 size = 3;
+    optional int64 timestamp = 4;
+    optional OffloadContext offloadContext = 5;
+    optional string bookieClusterName = 6;   // ★ new (Topic Data Ledger attribution)
+    repeated KeyValue properties = 7;        // ★ new (extension slot)

Review Comment:
   we already have the field `properties` in PIP-404. And maybe we can also use the field to store `bookieClusterName`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["wolfstudy (via GitHub)" <gi...@apache.org>]

wolfstudy commented on code in PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#discussion_r3295826919


##########
pip/pip-477.md:
##########
@@ -0,0 +1,991 @@
+# PIP-477: Dynamic Bookie Cluster Switching for Brokers
+
+> **Revision 2 (Plugin-First Architecture)** — This revision reframes PIP-477
+> per the design feedback to build on top of the storage-layer pluggability
+> introduced by [PIP-384](./pip-384.md), the per-ledger metadata extension slot
+> introduced by [PIP-404](./pip-404.md), the `CustomCommandFactory` SPI introduced
+> by [PIP-201](./pip-201.md), and the existing `AdditionalServlet` SPI — instead
+> of adding a self-contained subsystem to Pulsar core.
+
+## Motivation
+
+Apache Pulsar's broker layer is bound to a fixed BookKeeper (Bookie) cluster
+at startup through the `bookkeeperMetadataServiceUri` configuration. This
+static binding becomes a bottleneck in several real-world operational
+scenarios:
+
+1. **Slow failure recovery** — When a Bookie cluster experiences capacity
+   exhaustion, sustained latency degradation, or non-recoverable failures, the
+   only mitigation today is to wait for the Bookie cluster to recover. In
+   elastic-storage deployments where one Bookie cluster backs 100+ broker
+   clusters, the blast radius is enormous.
+2. **No capacity escape hatch** — When a Bookie cluster approaches its
+   capacity ceiling, brokers cannot "spill over" to a fresh Bookie cluster.
+3. **No runtime re-targeting** — There is no supported runtime mechanism to
+   migrate brokers from one Bookie cluster to another without restart and full
+   data migration.
+4. **Data plane coupling** — Switching deals with the data plane: terabytes of
+   ledger bytes, ongoing publishes, ongoing acknowledgments, and live cursors.
+   A copy-everything-then-cutover approach is infeasible.
+
+This PIP proposes a **safe, ledger-attribution-driven framework** that allows
+operators to switch a broker cluster's underlying Bookie cluster at runtime,
+with **zero data-plane downtime** and **per-ledger routing** that preserves the
+readability of historical data on the old cluster.
+
+The framework enables:
+
+- **Zero-downtime switching** — publish and consume continue without interruption
+- **Per-ledger routing** — each ledger carries its own cluster attribution
+- **No bulk data migration** — Topic Data Ledgers age out naturally via TTL/Retention
+- **Targeted migration of small, long-lived metadata ledgers** — only Schema and
+  Cursor Ledgers are copied, preserving the original `ledgerId`
+- **Automatic rollback** within a configurable window
+- **Operator-driven, not auto-failover** — avoids split-brain
+
+## Design Principle: Build on PIP-384, Not Around It
+
+The feature is delivered as **a `ManagedLedgerStorage` implementation that
+holds multiple `BookkeeperManagedLedgerStorageClass` instances**, packaged as
+an out-of-tree NAR plugin. The framing PIP-384 establishes —
+*"a topic resolves to one storage class"* — is generalized here to:
+
+> *"A topic resolves to one **active** storage class at a time, with
+> **attribution-driven routing for previously-written ledgers**."*
+
+Concretely, this PIP:
+
+- Lives in a separate Maven module (`pulsar-bookie-cluster-switching`) and
+  ships as a NAR; the broker loads it via the existing `managedLedgerStorageClassName`
+  configuration hook (PIP-384).
+- Re-uses **`LedgerInfo.properties`** (PIP-404, tag 6) and
+  **`ManagedCursorInfo.cursorProperties`** (existing, tag 8) for per-ledger
+  cluster attribution. **Zero new proto fields** in `MLDataFormats.proto`.
+- Ships its REST surface via **`AdditionalServlet`**; ships its CLI via
+  **`CustomCommandFactory` NAR** (PIP-201). **Zero changes** to `pulsar-broker`
+  REST classes or `pulsar-admin` CLI core.
+- Requires only **three narrow SPI hooks** in core (Section "Required Core SPI
+  Additions"), each useful to *any* custom `ManagedLedgerStorage` author and
+  much smaller than inlining the full feature.
+- When the plugin NAR is not installed, broker behavior is **byte-for-byte
+  identical** to unmodified Pulsar; the ~99% of users that never switch BK
+  clusters pay zero cost in core surface area.
+
+## Goal
+
+Provide a runtime mechanism by which a broker cluster can be re-pointed from
+one Bookie cluster (`oldCluster`) to a new Bookie cluster (`newCluster`)
+without service interruption, while preserving the ability to read historical
+data that physically resides on `oldCluster`.
+
+### In Scope
+
+- A plugin-provided `ManagedLedgerStorage` implementation
+  (`MultiClusterManagedLedgerStorage`) that holds a
+  `Map<clusterName, BookkeeperManagedLedgerStorageClass>` and exposes the
+  currently active one.
+- Per-ledger cluster attribution stored in the **existing**
+  `LedgerInfo.properties` and `ManagedCursorInfo.cursorProperties` slots under
+  the reserved key `_pulsar.bookieClusterName`.
+- Per-broker-cluster registry of multiple Bookie clusters, stored in
+  Broker-ZK under a plugin-owned path.
+- A plugin-owned `BookieClusterReadRouter` that selects a BK client at
+  ledger-handle open time based on the attribution property.
+- A plugin-owned `BookieClusterSwitchOrchestrator` that drives
+  `BUILD → PROMOTE → CLEANUP` for Schema and Cursor Ledgers (migration
+  preserves the original `ledgerId` via `BookKeeper.asyncCreateLedgerAdv`).
+- Plugin-provided CLI (`pulsar-admin bookie-clusters …`) via PIP-201
+  `CustomCommandFactory`.
+- Plugin-provided REST API under `/admin/v2/bookie-clusters/*` via
+  `AdditionalServlet`.
+- Three narrow core SPI hooks (see Section "Required Core SPI Additions").
+
+### Out of Scope
+
+- **Topic Data Ledger migration** — not migrated; ages out via TTL/Retention.
+- **Automatic failure detection and switch triggering** — operator-initiated
+  to avoid split-brain.
+- **Cross-cluster (geo-replicated) coordination** — each broker cluster
+  switches independently.
+- **BookKeeper-cluster-side internal migration tools** — delegated to BookKeeper.
+- **Changing `metadataStoreUrl` (Broker-ZK)** — only the Bookie cluster is
+  switched; Broker-ZK remains the same throughout.
+- **New proto fields in `MLDataFormats.proto`** — reuses existing extension
+  slots from PIP-404 and `cursorProperties`.
+
+## High-Level Architecture
+
+```
+┌────────────────────────────────────────────────────────────────────────┐
+│ Pulsar core (unchanged + 3 narrow SPI hooks)                           │
+│                                                                         │
+│   ManagedLedgerStorage SPI            ─── PIP-384                       │
+│   ManagedLedgerStorageClass SPI       ─── PIP-384                       │
+│   LedgerInfo.properties (KV)          ─── PIP-404                       │
+│   ManagedCursorInfo.cursorProperties  ─── pre-existing                  │
+│   AdditionalServlet SPI               ─── pre-existing                  │
+│   CustomCommandFactory SPI            ─── PIP-201                       │
+│                                                                         │
+│   + ManagedLedgerConfig.activeBookKeeperSupplier         ←── NEW (S1)   │
+│   + ManagedLedgerConfig.bookKeeperResolver               ←── NEW (S2)   │
+│   + SchemaStorageBookKeeperProvider SPI                  ←── NEW (S3)   │
+└────────────────────────────┬───────────────────────────────────────────┘
+                             │ SPI / NAR loading
+┌────────────────────────────▼───────────────────────────────────────────┐
+│ pulsar-bookie-cluster-switching (NAR, opt-in)                          │
+│                                                                         │
+│   MultiClusterManagedLedgerStorage   impl ManagedLedgerStorage         │
+│     ├─ Map<name, BookkeeperManagedLedgerStorageClass>                  │
+│     ├─ activeClusterName (volatile)                                     │
+│     └─ getDefaultStorageClass() → active                               │
+│                                                                         │
+│   RoutingManagedLedgerFactory        wraps ManagedLedgerFactoryImpl    │
+│     └─ Per-ledger BK resolution from LedgerInfo.properties             │
+│                                                                         │
+│   RoutingSchemaBookKeeperProvider    impl SchemaStorageBookKeeperProvider│
+│     └─ Per-position BK resolution from SchemaLocator extension         │
+│                                                                         │
+│   BookieClusterConfigManager / ConfigWatcher                            │
+│     └─ Owns ZK path /admin/bookie-clusters/* (plugin-owned)            │
+│                                                                         │
+│   BookieClusterSwitchOrchestrator    BUILD → PROMOTE → CLEANUP          │
+│                                                                         │
+│   REST  /admin/v2/bookie-clusters/*  via AdditionalServlet              │
+│   CLI   pulsar-admin bookie-clusters via CustomCommandFactory NAR       │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Key Principles
+
+| Principle | Description |
+|-----------|-------------|
+| **Attribution as Single Source of Truth** | Every ledger at creation is "stamped" with the active cluster name inside its existing metadata properties. Reads are routed by that stamp. No history table, no boundary heuristics, no side znode. |
+| **Reuse Extension Slots, Don't Add Proto Fields** | `LedgerInfo.properties` (PIP-404) and `ManagedCursorInfo.cursorProperties` carry the attribution. Only `SchemaStorageFormat.proto` needs one tiny extension (Section "Required Core SPI Additions" S3). |
+| **Topic Data: Don't Migrate** | Topic Data Ledgers have TTL/Retention; let them age out. Switching does not move terabytes of data. |
+| **Schema / Cursor: Migrate, but Preserve `ledgerId`** | We use `asyncCreateLedgerAdv(ledgerId, …)` to recreate the **same `ledgerId`** in the new cluster, copy entries with **identical `entryId`**, then CAS only the attribution property. Business keys (`cursorsLedgerId`, `position.ledgerId`, `position.entryId`) are byte-for-byte unchanged in Broker-ZK. |
+| **Plugin-First, Opt-In** | When the NAR is absent or `managedLedgerStorageClassName` ≠ the multi-cluster class, the broker is byte-for-byte unmodified Pulsar. |
+| **Operator-Driven, Single Active Cluster** | At any moment exactly one cluster is `status=ACTIVE` in Broker-ZK. Auto-failover is rejected. |
+| **Broker-ZK is the Switch Source-of-Truth** | All cluster registrations and the switch directive live in Broker-ZK under a plugin-owned path. No Global-ZK, no external coordinator. |
+
+## Required Core SPI Additions
+
+These are the **only** changes to Pulsar core required by this PIP. Each is a
+narrow, generally-useful hook that any custom `ManagedLedgerStorage`
+implementation could leverage — not specific to BK switching.
+
+### S1. `ManagedLedgerConfig.activeBookKeeperSupplier`
+
+**Problem.** Today `ManagedLedgerImpl` captures the `BookKeeper` client once
+in its constructor (via `ManagedLedgerFactoryImpl.bookkeeperFactory`). When a
+custom storage class needs to hot-swap the underlying BK client (e.g. cluster
+switch), there is no clean injection point on the write path.
+
+**Hook.** Add an optional supplier:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Supplier<BookKeeper> activeBookKeeperSupplier;   // nullable
+
+public Supplier<BookKeeper> getActiveBookKeeperSupplier() { … }
+public ManagedLedgerConfig setActiveBookKeeperSupplier(Supplier<BookKeeper> s) { … }
+```
+
+`ManagedLedgerImpl.createLedgerAfterClosed`, `rollCurrentLedgerIfFull`, and
+`ManagedCursorImpl.doCreateNewMetadataLedger` resolve the BK client as:
+
+```java
+BookKeeper bk = config.getActiveBookKeeperSupplier() != null
+        ? config.getActiveBookKeeperSupplier().get()
+        : this.bookKeeper;     // backward-compatible default
+```
+
+**Scope.** Pure additive. When the supplier is unset (default), behavior is
+identical. ~30 lines of change in `managed-ledger/`.
+
+### S2. `ManagedLedgerConfig.bookKeeperResolver`
+
+**Problem.** Reads (and deletes) need a *per-ledger* resolution path. Today
+`ManagedLedgerImpl` uses the constructor-captured `bookKeeper` for every
+`asyncOpenLedger`/`asyncDeleteLedger`. After a switch, ledgers that physically
+live on the old cluster must be opened with the old client.
+
+**Hook.** Add an optional resolver function keyed on `LedgerInfo`:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Function<LedgerInfo, BookKeeper> bookKeeperResolver;   // nullable
+
+// And the analogous function for cursor metadata ledgers:
+private Function<ManagedCursorInfo, BookKeeper> cursorBookKeeperResolver;
+```
+
+All call sites that today reference `this.bookKeeper` to open or delete a
+specific ledger (the comprehensive list is in
+`ManagedLedgerImpl.getLedgerHandle`, `internalAsyncOpenCursor`,
+`asyncDeleteLedger`, `asyncDeleteFromBookKeeper`, …) become:
+
+```java
+BookKeeper bk = resolveBookKeeperForLedger(ledgerInfo);
+```
+
+Where:
+
+```java
+private BookKeeper resolveBookKeeperForLedger(LedgerInfo li) {
+    var resolver = config.getBookKeeperResolver();
+    return resolver != null ? resolver.apply(li) : this.bookKeeper;
+}
+```
+
+**Scope.** Pure additive; when resolvers are unset, behavior is identical.
+~80 lines of change in `managed-ledger/`, mostly mechanical substitutions.
+
+### S3. `SchemaStorageBookKeeperProvider` SPI
+
+**Problem.** `BookkeeperSchemaStorage` captures `this.bookKeeper` at
+`start()` time (line ~101). It directly opens schema ledgers via that single
+client. To support per-ledger routing we need to inject a resolver and to
+attribute each schema ledger to a cluster.
+
+**Hook (two parts).**
+
+1. **Add a tiny extension slot to `SchemaStorageFormat.proto`** — this is the
+   *only* proto change in this PIP:
+
+   ```protobuf
+   message PositionInfo {
+       required int64 ledgerId = 1;
+       required int64 entryId = 2;
+       repeated KeyValue properties = 3;   // ★ NEW — generic extension slot,
+                                           // mirrors PIP-404 LedgerInfo.properties
+   }
+
+   message KeyValue {
+       required string key = 1;
+       required string value = 2;
+   }
+   ```
+
+   This is the minimal possible change: a generic `properties` slot symmetric
+   to PIP-404. It does **not** mention bookie clusters; any future plugin can
+   use it.
+
+2. **Add a provider SPI for the schema storage BK client:**
+
+   ```java
+   // pulsar-broker/.../service/schema/SchemaStorageBookKeeperProvider.java
+   public interface SchemaStorageBookKeeperProvider {
+       /** Returns the BK client to use for the given schema position. */
+       BookKeeper resolve(PositionInfo position);
+       /** Returns the BK client to use for *creating* a new schema ledger. */
+       BookKeeper active();
+   }
+   ```
+
+   `BookkeeperSchemaStorage` resolves its provider as:
+
+   ```java
+   var provider = pulsar.getSchemaStorageBookKeeperProvider();   // nullable
+   BookKeeper bk = provider != null ? provider.resolve(position) : this.bookKeeper;
+   ```
+
+   The default (provider == null) preserves today's behavior exactly.
+
+**Scope.** One proto field (additive, `repeated` is forward/backward
+compatible in proto2) + one SPI interface + ~50 lines in
+`BookkeeperSchemaStorage`. No new REST endpoint, no CLI change.
+
+### Summary: Core Diff Footprint
+
+| Component | LOC | Risk |
+|-----------|-----|------|
+| `ManagedLedgerConfig` (S1, S2 setters/getters) | ~40 | Trivial additive |
+| `ManagedLedgerImpl` / `ManagedCursorImpl` resolver wiring | ~80 | Mechanical, well-scoped |
+| `SchemaStorageFormat.proto` `PositionInfo.properties` (S3) | 1 field | Additive, proto2-safe |
+| `SchemaStorageBookKeeperProvider` SPI + `BookkeeperSchemaStorage` wiring | ~50 | Additive |
+| **Total core diff** | **~170 LOC, 1 new optional proto field** | **Zero behavioral change when SPI unused** |
+
+Everything else lives in the plugin NAR.
+
+## Plugin Module: `pulsar-bookie-cluster-switching`
+
+A new top-level Maven module that builds a NAR (`pulsar-bookie-cluster-switching-<version>.nar`).
+
+### Module Layout
+
+```
+pulsar-bookie-cluster-switching/
+├─ src/main/java/.../
+│   ├─ storage/
+│   │   ├─ MultiClusterManagedLedgerStorage.java    (implements ManagedLedgerStorage)
+│   │   ├─ MultiClusterStorageClass.java            (implements BookkeeperManagedLedgerStorageClass)
+│   │   └─ RoutingManagedLedgerFactory.java
+│   ├─ schema/
+│   │   └─ RoutingSchemaBookKeeperProvider.java     (implements SchemaStorageBookKeeperProvider)
+│   ├─ registry/
+│   │   ├─ BookieClusterConfigManager.java
+│   │   └─ BookieClusterConfigWatcher.java
+│   ├─ orchestrator/
+│   │   ├─ BookieClusterSwitchOrchestrator.java
+│   │   ├─ LedgerCopyUtil.java
+│   │   └─ IdgenPrecheckService.java
+│   ├─ rest/
+│   │   └─ BookieClusterAdminServlet.java           (implements AdditionalServlet)
+│   └─ cli/
+│       └─ BookieClusterCommandFactory.java         (implements CustomCommandFactory)
+└─ src/main/resources/META-INF/services/
+    ├─ org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet
+    └─ org.apache.pulsar.admin.cli.extensions.CustomCommandFactory
+```
+
+### Wiring on Broker Start
+
+The plugin is activated by setting in `broker.conf`:
+
+```properties
+managedLedgerStorageClassName=org.apache.pulsar.ext.bookieswitch.MultiClusterManagedLedgerStorage
+additionalServlets=bookie-clusters-admin
+additionalServletDirectory=./plugins
+```
+
+Then `ManagedLedgerStorage.create()` (PIP-384 entry point) reflectively loads
+`MultiClusterManagedLedgerStorage`. Inside its `initialize()`:
+
+1. Reads its own configuration namespace (prefix `bookieClusterSwitch.*`).
+2. Reads `/admin/bookie-clusters/*` from Broker-ZK to discover registered
+   clusters and resolve the initial `activeClusterName` (tier-1 →
+   Broker-ZK `status=ACTIVE` cluster; tier-2 → optional
+   `bookieClusterSwitch.currentClusterName` hint; tier-3 → derive from
+   `bookkeeperMetadataServiceUri` and auto-register as `ACTIVE`).
+3. Builds one `BookkeeperManagedLedgerStorageClass` per registered cluster
+   (each owns its own `BookKeeper` client and `StatsProvider`).
+4. Constructs a `RoutingManagedLedgerFactory` that decorates a base
+   `ManagedLedgerFactoryImpl` and injects the per-ledger BK resolver via the
+   S1/S2 hooks above.
+5. Starts `BookieClusterConfigWatcher` to receive switch directives.
+6. (Origin-broker-only after a switch) starts the orchestrator.
+
+The plugin's `AdditionalServlet` instance is independently discovered by
+`AdditionalServlets.load(…)` in `WebService`; the plugin's
+`CustomCommandFactory` is independently discovered by `pulsar-admin`'s NAR
+class loader.
+
+## Detailed Design
+
+### Per-Ledger Attribution Encoding
+
+| Ledger type | proto container | Encoding | Key |
+|-------------|-----------------|----------|-----|
+| Topic Data | `ManagedLedgerInfo.LedgerInfo.properties` (PIP-404, tag 6) | `KeyValue` | `_pulsar.bookieClusterName` |
+| Cursor | `ManagedCursorInfo.cursorProperties` (existing, tag 8) | `StringProperty` | `_pulsar.bookieClusterName` |
+| Schema | `SchemaStorageFormat.PositionInfo.properties` (NEW, tag 3, S3) | `KeyValue` | `_pulsar.bookieClusterName` |
+
+The reserved key prefix `_pulsar.` is **owned by Pulsar core** to avoid
+collisions with plugin- or user-defined properties. Plugin code uses a
+constant:
+
+```java
+public static final String ATTR_KEY = "_pulsar.bookieClusterName";
+```
+
+**Compatibility.**
+
+- **proto2 forward-compat:** `repeated KeyValue properties` already exists in
+  `LedgerInfo` (PIP-404). For `PositionInfo` it's added with a new tag (3),
+  matching the symmetric pattern.
+- **Old broker reading new metadata:** ignores `properties` it doesn't
+  understand; preserves them via `UnknownFieldSet` on rewrite (already the
+  case for PIP-404).
+- **New broker reading old metadata (no attribution):** the plugin's resolver
+  falls back to the active cluster (which equals the original cluster for
+  pre-switch ledgers; correct).
+
+### Write-Path Stamping
+
+`RoutingManagedLedgerFactory` wraps the base factory. When `ManagedLedger`
+asks for a new ledger:
+
+```java
+// Pseudocode of the stamping path
+String activeCluster = multiClusterStorage.getActiveClusterName();
+BookKeeper bk = multiClusterStorage.getBookKeeper(activeCluster);    // via S1
+LedgerHandle lh = bk.asyncCreateLedger(...);
+
+// In the same metadata CAS that records the new LedgerInfo:

Review Comment:
   Good idea, I will clarify this further in the PIP.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["void-ptr974 (via GitHub)" <gi...@apache.org>]

void-ptr974 commented on PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#issuecomment-4543717548

   Thanks, the rationale for keeping cursor/schema migration symmetric makes sense.
   
   One related question: should preserving the exact same ledgerId on the target BK cluster be a hard requirement of the design, or mainly an implementation simplification?
   
   If the target BK cluster is not empty, or is already serving other topics/tenants, the source ledgerId may already exist there. Since different BK clusters normally have independent ledgerId allocation spaces, a same-ledgerId copy seems to require an additional invariant: the target cluster must either reserve/import those ledgerIds safely, or guarantee that they cannot collide with existing/future ledgerIds.
   
   An alternative would be to allow the copied ledger to get a new target ledgerId and explicitly persist a source-ledger/source-position → target-ledger/target-position mapping. That would make PROMOTE/ROLLBACK and schema/cursor lookup more complex, but it may make the feature applicable to a wider set of real deployments where the target BK cluster is not dedicated or empty.
   
   I’m not saying the PIP must choose this approach, but it may be worth documenting why the design intentionally requires same-ledgerId copy, and what assumptions are required for it to be safe.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["wolfstudy (via GitHub)" <gi...@apache.org>]

wolfstudy commented on code in PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#discussion_r3295849560


##########
pip/pip-477.md:
##########
@@ -0,0 +1,991 @@
+# PIP-477: Dynamic Bookie Cluster Switching for Brokers
+
+> **Revision 2 (Plugin-First Architecture)** — This revision reframes PIP-477
+> per the design feedback to build on top of the storage-layer pluggability
+> introduced by [PIP-384](./pip-384.md), the per-ledger metadata extension slot
+> introduced by [PIP-404](./pip-404.md), the `CustomCommandFactory` SPI introduced
+> by [PIP-201](./pip-201.md), and the existing `AdditionalServlet` SPI — instead
+> of adding a self-contained subsystem to Pulsar core.
+
+## Motivation
+
+Apache Pulsar's broker layer is bound to a fixed BookKeeper (Bookie) cluster
+at startup through the `bookkeeperMetadataServiceUri` configuration. This
+static binding becomes a bottleneck in several real-world operational
+scenarios:
+
+1. **Slow failure recovery** — When a Bookie cluster experiences capacity
+   exhaustion, sustained latency degradation, or non-recoverable failures, the
+   only mitigation today is to wait for the Bookie cluster to recover. In
+   elastic-storage deployments where one Bookie cluster backs 100+ broker
+   clusters, the blast radius is enormous.
+2. **No capacity escape hatch** — When a Bookie cluster approaches its
+   capacity ceiling, brokers cannot "spill over" to a fresh Bookie cluster.
+3. **No runtime re-targeting** — There is no supported runtime mechanism to
+   migrate brokers from one Bookie cluster to another without restart and full
+   data migration.
+4. **Data plane coupling** — Switching deals with the data plane: terabytes of
+   ledger bytes, ongoing publishes, ongoing acknowledgments, and live cursors.
+   A copy-everything-then-cutover approach is infeasible.
+
+This PIP proposes a **safe, ledger-attribution-driven framework** that allows
+operators to switch a broker cluster's underlying Bookie cluster at runtime,
+with **zero data-plane downtime** and **per-ledger routing** that preserves the
+readability of historical data on the old cluster.
+
+The framework enables:
+
+- **Zero-downtime switching** — publish and consume continue without interruption
+- **Per-ledger routing** — each ledger carries its own cluster attribution
+- **No bulk data migration** — Topic Data Ledgers age out naturally via TTL/Retention
+- **Targeted migration of small, long-lived metadata ledgers** — only Schema and
+  Cursor Ledgers are copied, preserving the original `ledgerId`
+- **Automatic rollback** within a configurable window
+- **Operator-driven, not auto-failover** — avoids split-brain
+
+## Design Principle: Build on PIP-384, Not Around It
+
+The feature is delivered as **a `ManagedLedgerStorage` implementation that
+holds multiple `BookkeeperManagedLedgerStorageClass` instances**, packaged as
+an out-of-tree NAR plugin. The framing PIP-384 establishes —
+*"a topic resolves to one storage class"* — is generalized here to:
+
+> *"A topic resolves to one **active** storage class at a time, with
+> **attribution-driven routing for previously-written ledgers**."*
+
+Concretely, this PIP:
+
+- Lives in a separate Maven module (`pulsar-bookie-cluster-switching`) and
+  ships as a NAR; the broker loads it via the existing `managedLedgerStorageClassName`
+  configuration hook (PIP-384).
+- Re-uses **`LedgerInfo.properties`** (PIP-404, tag 6) and
+  **`ManagedCursorInfo.cursorProperties`** (existing, tag 8) for per-ledger
+  cluster attribution. **Zero new proto fields** in `MLDataFormats.proto`.
+- Ships its REST surface via **`AdditionalServlet`**; ships its CLI via
+  **`CustomCommandFactory` NAR** (PIP-201). **Zero changes** to `pulsar-broker`
+  REST classes or `pulsar-admin` CLI core.
+- Requires only **three narrow SPI hooks** in core (Section "Required Core SPI
+  Additions"), each useful to *any* custom `ManagedLedgerStorage` author and
+  much smaller than inlining the full feature.
+- When the plugin NAR is not installed, broker behavior is **byte-for-byte
+  identical** to unmodified Pulsar; the ~99% of users that never switch BK
+  clusters pay zero cost in core surface area.
+
+## Goal
+
+Provide a runtime mechanism by which a broker cluster can be re-pointed from
+one Bookie cluster (`oldCluster`) to a new Bookie cluster (`newCluster`)
+without service interruption, while preserving the ability to read historical
+data that physically resides on `oldCluster`.
+
+### In Scope
+
+- A plugin-provided `ManagedLedgerStorage` implementation
+  (`MultiClusterManagedLedgerStorage`) that holds a
+  `Map<clusterName, BookkeeperManagedLedgerStorageClass>` and exposes the
+  currently active one.
+- Per-ledger cluster attribution stored in the **existing**
+  `LedgerInfo.properties` and `ManagedCursorInfo.cursorProperties` slots under
+  the reserved key `_pulsar.bookieClusterName`.
+- Per-broker-cluster registry of multiple Bookie clusters, stored in
+  Broker-ZK under a plugin-owned path.
+- A plugin-owned `BookieClusterReadRouter` that selects a BK client at
+  ledger-handle open time based on the attribution property.
+- A plugin-owned `BookieClusterSwitchOrchestrator` that drives
+  `BUILD → PROMOTE → CLEANUP` for Schema and Cursor Ledgers (migration
+  preserves the original `ledgerId` via `BookKeeper.asyncCreateLedgerAdv`).
+- Plugin-provided CLI (`pulsar-admin bookie-clusters …`) via PIP-201
+  `CustomCommandFactory`.
+- Plugin-provided REST API under `/admin/v2/bookie-clusters/*` via
+  `AdditionalServlet`.
+- Three narrow core SPI hooks (see Section "Required Core SPI Additions").
+
+### Out of Scope
+
+- **Topic Data Ledger migration** — not migrated; ages out via TTL/Retention.
+- **Automatic failure detection and switch triggering** — operator-initiated
+  to avoid split-brain.
+- **Cross-cluster (geo-replicated) coordination** — each broker cluster
+  switches independently.
+- **BookKeeper-cluster-side internal migration tools** — delegated to BookKeeper.
+- **Changing `metadataStoreUrl` (Broker-ZK)** — only the Bookie cluster is
+  switched; Broker-ZK remains the same throughout.
+- **New proto fields in `MLDataFormats.proto`** — reuses existing extension
+  slots from PIP-404 and `cursorProperties`.
+
+## High-Level Architecture
+
+```
+┌────────────────────────────────────────────────────────────────────────┐
+│ Pulsar core (unchanged + 3 narrow SPI hooks)                           │
+│                                                                         │
+│   ManagedLedgerStorage SPI            ─── PIP-384                       │
+│   ManagedLedgerStorageClass SPI       ─── PIP-384                       │
+│   LedgerInfo.properties (KV)          ─── PIP-404                       │
+│   ManagedCursorInfo.cursorProperties  ─── pre-existing                  │
+│   AdditionalServlet SPI               ─── pre-existing                  │
+│   CustomCommandFactory SPI            ─── PIP-201                       │
+│                                                                         │
+│   + ManagedLedgerConfig.activeBookKeeperSupplier         ←── NEW (S1)   │
+│   + ManagedLedgerConfig.bookKeeperResolver               ←── NEW (S2)   │
+│   + SchemaStorageBookKeeperProvider SPI                  ←── NEW (S3)   │
+└────────────────────────────┬───────────────────────────────────────────┘
+                             │ SPI / NAR loading
+┌────────────────────────────▼───────────────────────────────────────────┐
+│ pulsar-bookie-cluster-switching (NAR, opt-in)                          │
+│                                                                         │
+│   MultiClusterManagedLedgerStorage   impl ManagedLedgerStorage         │
+│     ├─ Map<name, BookkeeperManagedLedgerStorageClass>                  │
+│     ├─ activeClusterName (volatile)                                     │
+│     └─ getDefaultStorageClass() → active                               │
+│                                                                         │
+│   RoutingManagedLedgerFactory        wraps ManagedLedgerFactoryImpl    │
+│     └─ Per-ledger BK resolution from LedgerInfo.properties             │
+│                                                                         │
+│   RoutingSchemaBookKeeperProvider    impl SchemaStorageBookKeeperProvider│
+│     └─ Per-position BK resolution from SchemaLocator extension         │
+│                                                                         │
+│   BookieClusterConfigManager / ConfigWatcher                            │
+│     └─ Owns ZK path /admin/bookie-clusters/* (plugin-owned)            │
+│                                                                         │
+│   BookieClusterSwitchOrchestrator    BUILD → PROMOTE → CLEANUP          │
+│                                                                         │
+│   REST  /admin/v2/bookie-clusters/*  via AdditionalServlet              │
+│   CLI   pulsar-admin bookie-clusters via CustomCommandFactory NAR       │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Key Principles
+
+| Principle | Description |
+|-----------|-------------|
+| **Attribution as Single Source of Truth** | Every ledger at creation is "stamped" with the active cluster name inside its existing metadata properties. Reads are routed by that stamp. No history table, no boundary heuristics, no side znode. |
+| **Reuse Extension Slots, Don't Add Proto Fields** | `LedgerInfo.properties` (PIP-404) and `ManagedCursorInfo.cursorProperties` carry the attribution. Only `SchemaStorageFormat.proto` needs one tiny extension (Section "Required Core SPI Additions" S3). |
+| **Topic Data: Don't Migrate** | Topic Data Ledgers have TTL/Retention; let them age out. Switching does not move terabytes of data. |
+| **Schema / Cursor: Migrate, but Preserve `ledgerId`** | We use `asyncCreateLedgerAdv(ledgerId, …)` to recreate the **same `ledgerId`** in the new cluster, copy entries with **identical `entryId`**, then CAS only the attribution property. Business keys (`cursorsLedgerId`, `position.ledgerId`, `position.entryId`) are byte-for-byte unchanged in Broker-ZK. |
+| **Plugin-First, Opt-In** | When the NAR is absent or `managedLedgerStorageClassName` ≠ the multi-cluster class, the broker is byte-for-byte unmodified Pulsar. |
+| **Operator-Driven, Single Active Cluster** | At any moment exactly one cluster is `status=ACTIVE` in Broker-ZK. Auto-failover is rejected. |
+| **Broker-ZK is the Switch Source-of-Truth** | All cluster registrations and the switch directive live in Broker-ZK under a plugin-owned path. No Global-ZK, no external coordinator. |
+
+## Required Core SPI Additions
+
+These are the **only** changes to Pulsar core required by this PIP. Each is a
+narrow, generally-useful hook that any custom `ManagedLedgerStorage`
+implementation could leverage — not specific to BK switching.
+
+### S1. `ManagedLedgerConfig.activeBookKeeperSupplier`
+
+**Problem.** Today `ManagedLedgerImpl` captures the `BookKeeper` client once
+in its constructor (via `ManagedLedgerFactoryImpl.bookkeeperFactory`). When a
+custom storage class needs to hot-swap the underlying BK client (e.g. cluster
+switch), there is no clean injection point on the write path.
+
+**Hook.** Add an optional supplier:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Supplier<BookKeeper> activeBookKeeperSupplier;   // nullable
+
+public Supplier<BookKeeper> getActiveBookKeeperSupplier() { … }
+public ManagedLedgerConfig setActiveBookKeeperSupplier(Supplier<BookKeeper> s) { … }
+```
+
+`ManagedLedgerImpl.createLedgerAfterClosed`, `rollCurrentLedgerIfFull`, and
+`ManagedCursorImpl.doCreateNewMetadataLedger` resolve the BK client as:
+
+```java
+BookKeeper bk = config.getActiveBookKeeperSupplier() != null
+        ? config.getActiveBookKeeperSupplier().get()
+        : this.bookKeeper;     // backward-compatible default
+```
+
+**Scope.** Pure additive. When the supplier is unset (default), behavior is
+identical. ~30 lines of change in `managed-ledger/`.
+
+### S2. `ManagedLedgerConfig.bookKeeperResolver`
+
+**Problem.** Reads (and deletes) need a *per-ledger* resolution path. Today
+`ManagedLedgerImpl` uses the constructor-captured `bookKeeper` for every
+`asyncOpenLedger`/`asyncDeleteLedger`. After a switch, ledgers that physically
+live on the old cluster must be opened with the old client.
+
+**Hook.** Add an optional resolver function keyed on `LedgerInfo`:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Function<LedgerInfo, BookKeeper> bookKeeperResolver;   // nullable
+
+// And the analogous function for cursor metadata ledgers:
+private Function<ManagedCursorInfo, BookKeeper> cursorBookKeeperResolver;
+```
+
+All call sites that today reference `this.bookKeeper` to open or delete a
+specific ledger (the comprehensive list is in
+`ManagedLedgerImpl.getLedgerHandle`, `internalAsyncOpenCursor`,
+`asyncDeleteLedger`, `asyncDeleteFromBookKeeper`, …) become:
+
+```java
+BookKeeper bk = resolveBookKeeperForLedger(ledgerInfo);
+```
+
+Where:
+
+```java
+private BookKeeper resolveBookKeeperForLedger(LedgerInfo li) {
+    var resolver = config.getBookKeeperResolver();
+    return resolver != null ? resolver.apply(li) : this.bookKeeper;
+}
+```
+
+**Scope.** Pure additive; when resolvers are unset, behavior is identical.
+~80 lines of change in `managed-ledger/`, mostly mechanical substitutions.
+
+### S3. `SchemaStorageBookKeeperProvider` SPI
+
+**Problem.** `BookkeeperSchemaStorage` captures `this.bookKeeper` at
+`start()` time (line ~101). It directly opens schema ledgers via that single
+client. To support per-ledger routing we need to inject a resolver and to
+attribute each schema ledger to a cluster.
+
+**Hook (two parts).**
+
+1. **Add a tiny extension slot to `SchemaStorageFormat.proto`** — this is the
+   *only* proto change in this PIP:
+
+   ```protobuf
+   message PositionInfo {
+       required int64 ledgerId = 1;
+       required int64 entryId = 2;
+       repeated KeyValue properties = 3;   // ★ NEW — generic extension slot,
+                                           // mirrors PIP-404 LedgerInfo.properties
+   }
+
+   message KeyValue {
+       required string key = 1;
+       required string value = 2;
+   }
+   ```
+
+   This is the minimal possible change: a generic `properties` slot symmetric
+   to PIP-404. It does **not** mention bookie clusters; any future plugin can
+   use it.
+
+2. **Add a provider SPI for the schema storage BK client:**
+
+   ```java
+   // pulsar-broker/.../service/schema/SchemaStorageBookKeeperProvider.java
+   public interface SchemaStorageBookKeeperProvider {
+       /** Returns the BK client to use for the given schema position. */
+       BookKeeper resolve(PositionInfo position);
+       /** Returns the BK client to use for *creating* a new schema ledger. */
+       BookKeeper active();
+   }
+   ```
+
+   `BookkeeperSchemaStorage` resolves its provider as:
+
+   ```java
+   var provider = pulsar.getSchemaStorageBookKeeperProvider();   // nullable
+   BookKeeper bk = provider != null ? provider.resolve(position) : this.bookKeeper;
+   ```
+
+   The default (provider == null) preserves today's behavior exactly.
+
+**Scope.** One proto field (additive, `repeated` is forward/backward
+compatible in proto2) + one SPI interface + ~50 lines in
+`BookkeeperSchemaStorage`. No new REST endpoint, no CLI change.
+
+### Summary: Core Diff Footprint
+
+| Component | LOC | Risk |
+|-----------|-----|------|
+| `ManagedLedgerConfig` (S1, S2 setters/getters) | ~40 | Trivial additive |
+| `ManagedLedgerImpl` / `ManagedCursorImpl` resolver wiring | ~80 | Mechanical, well-scoped |
+| `SchemaStorageFormat.proto` `PositionInfo.properties` (S3) | 1 field | Additive, proto2-safe |
+| `SchemaStorageBookKeeperProvider` SPI + `BookkeeperSchemaStorage` wiring | ~50 | Additive |
+| **Total core diff** | **~170 LOC, 1 new optional proto field** | **Zero behavioral change when SPI unused** |
+
+Everything else lives in the plugin NAR.
+
+## Plugin Module: `pulsar-bookie-cluster-switching`
+
+A new top-level Maven module that builds a NAR (`pulsar-bookie-cluster-switching-<version>.nar`).
+
+### Module Layout
+
+```
+pulsar-bookie-cluster-switching/
+├─ src/main/java/.../
+│   ├─ storage/
+│   │   ├─ MultiClusterManagedLedgerStorage.java    (implements ManagedLedgerStorage)
+│   │   ├─ MultiClusterStorageClass.java            (implements BookkeeperManagedLedgerStorageClass)
+│   │   └─ RoutingManagedLedgerFactory.java
+│   ├─ schema/
+│   │   └─ RoutingSchemaBookKeeperProvider.java     (implements SchemaStorageBookKeeperProvider)
+│   ├─ registry/
+│   │   ├─ BookieClusterConfigManager.java
+│   │   └─ BookieClusterConfigWatcher.java
+│   ├─ orchestrator/
+│   │   ├─ BookieClusterSwitchOrchestrator.java
+│   │   ├─ LedgerCopyUtil.java
+│   │   └─ IdgenPrecheckService.java
+│   ├─ rest/
+│   │   └─ BookieClusterAdminServlet.java           (implements AdditionalServlet)
+│   └─ cli/
+│       └─ BookieClusterCommandFactory.java         (implements CustomCommandFactory)
+└─ src/main/resources/META-INF/services/
+    ├─ org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet
+    └─ org.apache.pulsar.admin.cli.extensions.CustomCommandFactory
+```
+
+### Wiring on Broker Start
+
+The plugin is activated by setting in `broker.conf`:
+
+```properties
+managedLedgerStorageClassName=org.apache.pulsar.ext.bookieswitch.MultiClusterManagedLedgerStorage
+additionalServlets=bookie-clusters-admin
+additionalServletDirectory=./plugins
+```
+
+Then `ManagedLedgerStorage.create()` (PIP-384 entry point) reflectively loads
+`MultiClusterManagedLedgerStorage`. Inside its `initialize()`:
+
+1. Reads its own configuration namespace (prefix `bookieClusterSwitch.*`).
+2. Reads `/admin/bookie-clusters/*` from Broker-ZK to discover registered
+   clusters and resolve the initial `activeClusterName` (tier-1 →
+   Broker-ZK `status=ACTIVE` cluster; tier-2 → optional
+   `bookieClusterSwitch.currentClusterName` hint; tier-3 → derive from
+   `bookkeeperMetadataServiceUri` and auto-register as `ACTIVE`).
+3. Builds one `BookkeeperManagedLedgerStorageClass` per registered cluster
+   (each owns its own `BookKeeper` client and `StatsProvider`).
+4. Constructs a `RoutingManagedLedgerFactory` that decorates a base
+   `ManagedLedgerFactoryImpl` and injects the per-ledger BK resolver via the
+   S1/S2 hooks above.
+5. Starts `BookieClusterConfigWatcher` to receive switch directives.
+6. (Origin-broker-only after a switch) starts the orchestrator.
+
+The plugin's `AdditionalServlet` instance is independently discovered by
+`AdditionalServlets.load(…)` in `WebService`; the plugin's
+`CustomCommandFactory` is independently discovered by `pulsar-admin`'s NAR
+class loader.
+
+## Detailed Design
+
+### Per-Ledger Attribution Encoding
+
+| Ledger type | proto container | Encoding | Key |
+|-------------|-----------------|----------|-----|
+| Topic Data | `ManagedLedgerInfo.LedgerInfo.properties` (PIP-404, tag 6) | `KeyValue` | `_pulsar.bookieClusterName` |
+| Cursor | `ManagedCursorInfo.cursorProperties` (existing, tag 8) | `StringProperty` | `_pulsar.bookieClusterName` |
+| Schema | `SchemaStorageFormat.PositionInfo.properties` (NEW, tag 3, S3) | `KeyValue` | `_pulsar.bookieClusterName` |
+
+The reserved key prefix `_pulsar.` is **owned by Pulsar core** to avoid
+collisions with plugin- or user-defined properties. Plugin code uses a
+constant:
+
+```java
+public static final String ATTR_KEY = "_pulsar.bookieClusterName";
+```
+
+**Compatibility.**
+
+- **proto2 forward-compat:** `repeated KeyValue properties` already exists in
+  `LedgerInfo` (PIP-404). For `PositionInfo` it's added with a new tag (3),
+  matching the symmetric pattern.
+- **Old broker reading new metadata:** ignores `properties` it doesn't
+  understand; preserves them via `UnknownFieldSet` on rewrite (already the
+  case for PIP-404).
+- **New broker reading old metadata (no attribution):** the plugin's resolver
+  falls back to the active cluster (which equals the original cluster for
+  pre-switch ledgers; correct).
+
+### Write-Path Stamping
+
+`RoutingManagedLedgerFactory` wraps the base factory. When `ManagedLedger`
+asks for a new ledger:
+
+```java
+// Pseudocode of the stamping path
+String activeCluster = multiClusterStorage.getActiveClusterName();
+BookKeeper bk = multiClusterStorage.getBookKeeper(activeCluster);    // via S1
+LedgerHandle lh = bk.asyncCreateLedger(...);
+
+// In the same metadata CAS that records the new LedgerInfo:
+LedgerInfo info = baseInfo.toBuilder()
+    .addProperties(KeyValue.newBuilder()
+        .setKey(ATTR_KEY)
+        .setValue(activeCluster)
+        .build())
+    .build();
+```
+
+The same pattern applies to:
+
+- `ManagedCursorImpl.doCreateNewMetadataLedger` → stamps
+  `cursorProperties[_pulsar.bookieClusterName] = activeCluster`.
+- `BookkeeperSchemaStorage.createLedger` (via S3 provider) → stamps
+  `PositionInfo.properties[_pulsar.bookieClusterName] = activeCluster`.
+
+The stamp is written in the **same Broker-ZK CAS** as the business field
+(`ledgerId`, `cursorsLedgerId`, `position`), guaranteeing atomicity
+(invariant **I1**).
+
+### Read-Path Routing
+
+```java
+// RoutingManagedLedgerFactory -- injected as ManagedLedgerConfig.bookKeeperResolver (S2)
+BookKeeper resolve(LedgerInfo li) {
+    String cluster = li.getPropertiesList().stream()
+        .filter(kv -> ATTR_KEY.equals(kv.getKey()))
+        .map(KeyValue::getValue)
+        .findFirst()
+        .orElseGet(multiClusterStorage::getActiveClusterName);   // legacy fallback
+    return multiClusterStorage.getBookKeeper(cluster);
+}
+```
+
+Identical pattern for cursors (consulting `cursorProperties`) and schema
+positions (consulting `PositionInfo.properties`).
+
+**Invariant I4** (entry-level consistency within a single read): the resolver
+is consulted exactly once at `LedgerHandle` open time; the resulting
+`LedgerHandle` is bound to one BK client for the entire read of that ledger.
+
+### Switch Phases (Operator-Driven State Machine)
+
+```
+NOT_REGISTERED
+    ↓
+[Operator: pulsar-admin bookie-clusters register --name newCluster --uri … --status STANDBY]
+    ↓
+REGISTERED (STANDBY)
+    ↓
+[Operator: pulsar-admin bookie-clusters switch --target newCluster]
+    ↓ (inline precheck: idgen-long advanced)
+SWITCH_TRIGGERED ← Coordinator broker writes /admin/bookie-clusters/switch-target
+    ↓
+LIVE_DUAL_READ  ← All brokers' watchers fire; new writes stamped newCluster;
+    ↓             reads of pre-stamped ledgers routed by attribution
+BUILD_PROMOTE_CLEANUP
+    ├── BUILD     ← Coordinator copies Schema/Cursor ledger bytes to newCluster
+    │             with the SAME ledgerId; entries copied with the SAME entryId
+    ├── PROMOTE   ← Schema: coordinator CAS-rewrites SchemaLocator
+    │             (changes only PositionInfo.properties[ATTR_KEY])
+    │             Cursor: coordinator forwards POST /internal/promote-cursor
+    │             to topic owner; owner CAS in ManagedCursorImpl lock
+    └── CLEANUP   ← After rollback window, delete old-cluster copies; promote
+                  Broker-ZK status: oldCluster→DEPRECATED, newCluster→ACTIVE
+    ↓
+DONE / DONE_WITH_FAILURES
+
+(Operator-triggered, within window):
+ROLLBACK ← Reverse-CAS attribution back to oldCluster; new-cluster
+          copies scheduled for delayed deletion.
+```
+
+All states above live **inside the plugin**. The orchestrator persists its
+progress under `/admin/bookie-clusters/orchestrator/<switchId>/...`.
+
+### Phase 1: Registration
+
+```bash
+pulsar-admin bookie-clusters register \
+    --name new-cluster \
+    --metadata-service-uri zk+null://new-zk:2181/ledgers-v2 \
+    --status STANDBY
+```
+
+`BookieClusterConfigManager` (plugin) validates:
+
+- `metadata-service-uri` does not collide with any already-registered cluster.
+- For co-located deployments, the chroot must differ from existing clusters.
+
+### Phase 2: Precheck (Invariant I11)
+
+```bash
+pulsar-admin bookie-clusters precheck --name new-cluster
+```
+
+`IdgenPrecheckService` (plugin) verifies the new cluster's `/ledgers/idgen-long`
+is advanced beyond `max(ledgerId)` of the source cluster. Otherwise
+`asyncCreateLedgerAdv(ledgerId, …)` during BUILD would collide with
+newly-allocated ledgerIds. `POST /switch` invokes precheck inline; failure
+returns HTTP 409 (bypassable via plugin config in lab/staging only).
+
+### Phase 3: Switch Trigger
+
+```bash
+pulsar-admin bookie-clusters switch --target new-cluster
+```
+
+The receiving broker (origin) writes:
+
+```
+/admin/bookie-clusters/switch-target
+  body: { "target": "new-cluster", "origin": "https://broker-1:8080",
+          "epoch": 7, "ts": 1716256800000 }
+```
+
+All brokers' `BookieClusterConfigWatcher` fire and execute, in order:
+
+1. `MultiClusterManagedLedgerStorage.switchActiveCluster(newCluster)` —
+   hot-swap `activeClusterName`; new BK client lazily created if not present.
+2. First switch: enable dual-read in the read router; subsequent switches:
+   refresh epoch.
+3. Emit metrics.
+4. **Origin broker only:** start the orchestrator (`BUILD → PROMOTE → CLEANUP`).
+   Non-origin brokers do not start an orchestrator. If the directive body
+   lacks `origin`, brokers fall back to leader election among themselves
+   (using the existing `LeaderElectionService`) to pick exactly one
+   orchestrator.
+
+### Phase 4: BUILD (Schema + Cursor Ledger Bytes)
+
+The orchestrator runs **two parallel stages** with bounded concurrency.
+
+#### Schema Ledger BUILD (centralized)
+
+```
+for each schemaId in /schemas/* (parallel, bounded):
+  locator = brokerZk.get("/schemas/<schemaId>")
+  for each indexEntry in locator.indexList:
+    ledgerId = indexEntry.position.ledgerId
+    cluster  = readClusterAttr(indexEntry.position) ?? activeClusterAtFirstWrite
+    if cluster == oldCluster:
+      LedgerCopyUtil.copyLedgerPreservingIds(
+          ledgerId, oldBk, newBk,
+          customMetadata,           // identical to source ledger
+          ensembleSize, qw, qa)
+        // Internally:
+        //   newBk.asyncCreateLedgerAdv(ledgerId, …)        ← SAME ledgerId
+        //   for entryId in [0, lastAddConfirmed]:
+        //     newLh.addEntry(entryId, srcEntry)            ← SAME entryId
+        //   newLh.close()
+```
+
+#### Cursor Ledger BUILD (centralized)
+
+Identical pattern over `/managed-ledgers/<topic>/<cursorName>` znodes, copying

Review Comment:
   > Could we discuss whether cursor metadata needs to be migrated by copying the existing cursor ledger with the same ledgerId? If the topic owner can force a cursor metadata checkpoint/rollover after the active BK cluster changes, ... would that be sufficient ... ?
   
   Thanks — this is a sharp observation and you are right that, considered in isolation, a cursor metadata ledger is semantically just a checkpoint: only the latest cursorsLedgerId matters; older cursor ledgers carry no business value once superseded. So technically we could indeed handle cursors with a forced rollover instead of byte-copying with asyncCreateLedgerAdv.
   
   We have considered this and chose to keep the byte-copy + ledgerId-preserving design for cursors. Our reasoning, for the record (and to make the trade-off auditable), is now captured explicitly in the PIP under **"Alternatives Considered → Alt-5"**. The short version:
   
   1. Symmetry with Schema is load-bearing. Schema metadata ledgers are referenced directly by SchemaLocator.indexList[].position.ledgerId/entryId for every historical-version read; there is no equivalent "rollover and forget" path for schema. If we adopted rollover for cursors, BUILD/PROMOTE would be two divergent state machines (one mutating cursorsLedgerId, one immutable on position.ledgerId). With the unified design, both paths are "byte-copy with same ledgerId → CAS only the attribution property", and LedgerCopyUtil is one shared utility.
   
   2. Rollback would no longer be O(1). Today's PROMOTE only flips _pulsar.bookieClusterName; ROLLBACK is the reverse flip — constant-time, no data movement (§"Phase 7"). Under rollover-based migration, cursorsLedgerId itself moves, so ROLLBACK either has to mint another fresh cursor ledger on the old cluster or resurrect a previously-deleted ledgerId. That's more code and more failure modes for an operation we want to keep boring.
   
   3. I9 / I8 invariants weaken. The current design's "PROMOTE changes ONLY the attribution KV" is auditable: a single CAS, one field. Alt-5 turns PROMOTE into a multi-field rewrite of ManagedCursorInfo, which couples the migration path with the markDelete-flush path on owner brokers.
   
   4. Owner-broker tail latency. Forced rollover must run under the ManagedCursorImpl lock, serialized with in-flight markDelete flushes and BK acks, before the CAS can proceed. Performing this concurrently across every cursor on every owner during the switch window has measurable impact on markDelete p99. Today's BUILD runs entirely on the orchestrator broker against dormant copies; owner brokers see only a sub-second PROMOTE RPC.
   
   5. The byte-copy cost is bounded. Cursor metadata ledgers are KB–MB; copying them with bounded concurrency on the orchestrator is cheap and pays for itself by simplifying every other corner of the design.
   
   So we are keeping the current cursor BUILD path. I've added Alt-5 to the PIP so this conversation is preserved and a future PIP can revisit the trade-off if priorities change.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#discussion_r3246980306


##########
pip/pip-470.md:
##########
@@ -0,0 +1,587 @@
+# PIP-470: Dynamic Bookie Cluster Switching for Brokers

Review Comment:
   PIP-470 is taken. PIP-477 is currently available. When the discussion thread is started on dev mailing list, that's how to reserve a PIP number.



##########
pip/pip-470.md:
##########
@@ -0,0 +1,587 @@
+# PIP-470: Dynamic Bookie Cluster Switching for Brokers

Review Comment:
   ```suggestion
   # PIP-477: Dynamic Bookie Cluster Switching for Brokers
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["void-ptr974 (via GitHub)" <gi...@apache.org>]

void-ptr974 commented on code in PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#discussion_r3303502497


##########
pip/pip-477.md:
##########
@@ -0,0 +1,991 @@
+# PIP-477: Dynamic Bookie Cluster Switching for Brokers
+
+> **Revision 2 (Plugin-First Architecture)** — This revision reframes PIP-477
+> per the design feedback to build on top of the storage-layer pluggability
+> introduced by [PIP-384](./pip-384.md), the per-ledger metadata extension slot
+> introduced by [PIP-404](./pip-404.md), the `CustomCommandFactory` SPI introduced
+> by [PIP-201](./pip-201.md), and the existing `AdditionalServlet` SPI — instead
+> of adding a self-contained subsystem to Pulsar core.
+
+## Motivation
+
+Apache Pulsar's broker layer is bound to a fixed BookKeeper (Bookie) cluster
+at startup through the `bookkeeperMetadataServiceUri` configuration. This
+static binding becomes a bottleneck in several real-world operational
+scenarios:
+
+1. **Slow failure recovery** — When a Bookie cluster experiences capacity
+   exhaustion, sustained latency degradation, or non-recoverable failures, the
+   only mitigation today is to wait for the Bookie cluster to recover. In
+   elastic-storage deployments where one Bookie cluster backs 100+ broker
+   clusters, the blast radius is enormous.
+2. **No capacity escape hatch** — When a Bookie cluster approaches its
+   capacity ceiling, brokers cannot "spill over" to a fresh Bookie cluster.
+3. **No runtime re-targeting** — There is no supported runtime mechanism to
+   migrate brokers from one Bookie cluster to another without restart and full
+   data migration.
+4. **Data plane coupling** — Switching deals with the data plane: terabytes of
+   ledger bytes, ongoing publishes, ongoing acknowledgments, and live cursors.
+   A copy-everything-then-cutover approach is infeasible.
+
+This PIP proposes a **safe, ledger-attribution-driven framework** that allows
+operators to switch a broker cluster's underlying Bookie cluster at runtime,
+with **zero data-plane downtime** and **per-ledger routing** that preserves the
+readability of historical data on the old cluster.
+
+The framework enables:
+
+- **Zero-downtime switching** — publish and consume continue without interruption
+- **Per-ledger routing** — each ledger carries its own cluster attribution
+- **No bulk data migration** — Topic Data Ledgers age out naturally via TTL/Retention
+- **Targeted migration of small, long-lived metadata ledgers** — only Schema and
+  Cursor Ledgers are copied, preserving the original `ledgerId`
+- **Automatic rollback** within a configurable window
+- **Operator-driven, not auto-failover** — avoids split-brain
+
+## Design Principle: Build on PIP-384, Not Around It
+
+The feature is delivered as **a `ManagedLedgerStorage` implementation that
+holds multiple `BookkeeperManagedLedgerStorageClass` instances**, packaged as
+an out-of-tree NAR plugin. The framing PIP-384 establishes —
+*"a topic resolves to one storage class"* — is generalized here to:
+
+> *"A topic resolves to one **active** storage class at a time, with
+> **attribution-driven routing for previously-written ledgers**."*
+
+Concretely, this PIP:
+
+- Lives in a separate Maven module (`pulsar-bookie-cluster-switching`) and
+  ships as a NAR; the broker loads it via the existing `managedLedgerStorageClassName`
+  configuration hook (PIP-384).
+- Re-uses **`LedgerInfo.properties`** (PIP-404, tag 6) and
+  **`ManagedCursorInfo.cursorProperties`** (existing, tag 8) for per-ledger
+  cluster attribution. **Zero new proto fields** in `MLDataFormats.proto`.
+- Ships its REST surface via **`AdditionalServlet`**; ships its CLI via
+  **`CustomCommandFactory` NAR** (PIP-201). **Zero changes** to `pulsar-broker`
+  REST classes or `pulsar-admin` CLI core.
+- Requires only **three narrow SPI hooks** in core (Section "Required Core SPI
+  Additions"), each useful to *any* custom `ManagedLedgerStorage` author and
+  much smaller than inlining the full feature.
+- When the plugin NAR is not installed, broker behavior is **byte-for-byte
+  identical** to unmodified Pulsar; the ~99% of users that never switch BK
+  clusters pay zero cost in core surface area.
+
+## Goal
+
+Provide a runtime mechanism by which a broker cluster can be re-pointed from
+one Bookie cluster (`oldCluster`) to a new Bookie cluster (`newCluster`)
+without service interruption, while preserving the ability to read historical
+data that physically resides on `oldCluster`.
+
+### In Scope
+
+- A plugin-provided `ManagedLedgerStorage` implementation
+  (`MultiClusterManagedLedgerStorage`) that holds a
+  `Map<clusterName, BookkeeperManagedLedgerStorageClass>` and exposes the
+  currently active one.
+- Per-ledger cluster attribution stored in the **existing**
+  `LedgerInfo.properties` and `ManagedCursorInfo.cursorProperties` slots under
+  the reserved key `_pulsar.bookieClusterName`.
+- Per-broker-cluster registry of multiple Bookie clusters, stored in
+  Broker-ZK under a plugin-owned path.
+- A plugin-owned `BookieClusterReadRouter` that selects a BK client at
+  ledger-handle open time based on the attribution property.
+- A plugin-owned `BookieClusterSwitchOrchestrator` that drives
+  `BUILD → PROMOTE → CLEANUP` for Schema and Cursor Ledgers (migration
+  preserves the original `ledgerId` via `BookKeeper.asyncCreateLedgerAdv`).
+- Plugin-provided CLI (`pulsar-admin bookie-clusters …`) via PIP-201
+  `CustomCommandFactory`.
+- Plugin-provided REST API under `/admin/v2/bookie-clusters/*` via
+  `AdditionalServlet`.
+- Three narrow core SPI hooks (see Section "Required Core SPI Additions").
+
+### Out of Scope
+
+- **Topic Data Ledger migration** — not migrated; ages out via TTL/Retention.
+- **Automatic failure detection and switch triggering** — operator-initiated
+  to avoid split-brain.
+- **Cross-cluster (geo-replicated) coordination** — each broker cluster
+  switches independently.
+- **BookKeeper-cluster-side internal migration tools** — delegated to BookKeeper.
+- **Changing `metadataStoreUrl` (Broker-ZK)** — only the Bookie cluster is
+  switched; Broker-ZK remains the same throughout.
+- **New proto fields in `MLDataFormats.proto`** — reuses existing extension
+  slots from PIP-404 and `cursorProperties`.
+
+## High-Level Architecture
+
+```
+┌────────────────────────────────────────────────────────────────────────┐
+│ Pulsar core (unchanged + 3 narrow SPI hooks)                           │
+│                                                                         │
+│   ManagedLedgerStorage SPI            ─── PIP-384                       │
+│   ManagedLedgerStorageClass SPI       ─── PIP-384                       │
+│   LedgerInfo.properties (KV)          ─── PIP-404                       │
+│   ManagedCursorInfo.cursorProperties  ─── pre-existing                  │
+│   AdditionalServlet SPI               ─── pre-existing                  │
+│   CustomCommandFactory SPI            ─── PIP-201                       │
+│                                                                         │
+│   + ManagedLedgerConfig.activeBookKeeperSupplier         ←── NEW (S1)   │
+│   + ManagedLedgerConfig.bookKeeperResolver               ←── NEW (S2)   │
+│   + SchemaStorageBookKeeperProvider SPI                  ←── NEW (S3)   │
+└────────────────────────────┬───────────────────────────────────────────┘
+                             │ SPI / NAR loading
+┌────────────────────────────▼───────────────────────────────────────────┐
+│ pulsar-bookie-cluster-switching (NAR, opt-in)                          │
+│                                                                         │
+│   MultiClusterManagedLedgerStorage   impl ManagedLedgerStorage         │
+│     ├─ Map<name, BookkeeperManagedLedgerStorageClass>                  │
+│     ├─ activeClusterName (volatile)                                     │
+│     └─ getDefaultStorageClass() → active                               │
+│                                                                         │
+│   RoutingManagedLedgerFactory        wraps ManagedLedgerFactoryImpl    │
+│     └─ Per-ledger BK resolution from LedgerInfo.properties             │
+│                                                                         │
+│   RoutingSchemaBookKeeperProvider    impl SchemaStorageBookKeeperProvider│
+│     └─ Per-position BK resolution from SchemaLocator extension         │
+│                                                                         │
+│   BookieClusterConfigManager / ConfigWatcher                            │
+│     └─ Owns ZK path /admin/bookie-clusters/* (plugin-owned)            │
+│                                                                         │
+│   BookieClusterSwitchOrchestrator    BUILD → PROMOTE → CLEANUP          │
+│                                                                         │
+│   REST  /admin/v2/bookie-clusters/*  via AdditionalServlet              │
+│   CLI   pulsar-admin bookie-clusters via CustomCommandFactory NAR       │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Key Principles
+
+| Principle | Description |
+|-----------|-------------|
+| **Attribution as Single Source of Truth** | Every ledger at creation is "stamped" with the active cluster name inside its existing metadata properties. Reads are routed by that stamp. No history table, no boundary heuristics, no side znode. |
+| **Reuse Extension Slots, Don't Add Proto Fields** | `LedgerInfo.properties` (PIP-404) and `ManagedCursorInfo.cursorProperties` carry the attribution. Only `SchemaStorageFormat.proto` needs one tiny extension (Section "Required Core SPI Additions" S3). |
+| **Topic Data: Don't Migrate** | Topic Data Ledgers have TTL/Retention; let them age out. Switching does not move terabytes of data. |
+| **Schema / Cursor: Migrate, but Preserve `ledgerId`** | We use `asyncCreateLedgerAdv(ledgerId, …)` to recreate the **same `ledgerId`** in the new cluster, copy entries with **identical `entryId`**, then CAS only the attribution property. Business keys (`cursorsLedgerId`, `position.ledgerId`, `position.entryId`) are byte-for-byte unchanged in Broker-ZK. |
+| **Plugin-First, Opt-In** | When the NAR is absent or `managedLedgerStorageClassName` ≠ the multi-cluster class, the broker is byte-for-byte unmodified Pulsar. |
+| **Operator-Driven, Single Active Cluster** | At any moment exactly one cluster is `status=ACTIVE` in Broker-ZK. Auto-failover is rejected. |
+| **Broker-ZK is the Switch Source-of-Truth** | All cluster registrations and the switch directive live in Broker-ZK under a plugin-owned path. No Global-ZK, no external coordinator. |
+
+## Required Core SPI Additions
+
+These are the **only** changes to Pulsar core required by this PIP. Each is a
+narrow, generally-useful hook that any custom `ManagedLedgerStorage`
+implementation could leverage — not specific to BK switching.
+
+### S1. `ManagedLedgerConfig.activeBookKeeperSupplier`
+
+**Problem.** Today `ManagedLedgerImpl` captures the `BookKeeper` client once
+in its constructor (via `ManagedLedgerFactoryImpl.bookkeeperFactory`). When a
+custom storage class needs to hot-swap the underlying BK client (e.g. cluster
+switch), there is no clean injection point on the write path.
+
+**Hook.** Add an optional supplier:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Supplier<BookKeeper> activeBookKeeperSupplier;   // nullable
+
+public Supplier<BookKeeper> getActiveBookKeeperSupplier() { … }
+public ManagedLedgerConfig setActiveBookKeeperSupplier(Supplier<BookKeeper> s) { … }
+```
+
+`ManagedLedgerImpl.createLedgerAfterClosed`, `rollCurrentLedgerIfFull`, and
+`ManagedCursorImpl.doCreateNewMetadataLedger` resolve the BK client as:
+
+```java
+BookKeeper bk = config.getActiveBookKeeperSupplier() != null
+        ? config.getActiveBookKeeperSupplier().get()
+        : this.bookKeeper;     // backward-compatible default
+```
+
+**Scope.** Pure additive. When the supplier is unset (default), behavior is
+identical. ~30 lines of change in `managed-ledger/`.
+
+### S2. `ManagedLedgerConfig.bookKeeperResolver`
+
+**Problem.** Reads (and deletes) need a *per-ledger* resolution path. Today
+`ManagedLedgerImpl` uses the constructor-captured `bookKeeper` for every
+`asyncOpenLedger`/`asyncDeleteLedger`. After a switch, ledgers that physically
+live on the old cluster must be opened with the old client.
+
+**Hook.** Add an optional resolver function keyed on `LedgerInfo`:
+
+```java
+// managed-ledger/.../ManagedLedgerConfig.java
+private Function<LedgerInfo, BookKeeper> bookKeeperResolver;   // nullable
+
+// And the analogous function for cursor metadata ledgers:
+private Function<ManagedCursorInfo, BookKeeper> cursorBookKeeperResolver;
+```
+
+All call sites that today reference `this.bookKeeper` to open or delete a
+specific ledger (the comprehensive list is in
+`ManagedLedgerImpl.getLedgerHandle`, `internalAsyncOpenCursor`,
+`asyncDeleteLedger`, `asyncDeleteFromBookKeeper`, …) become:
+
+```java
+BookKeeper bk = resolveBookKeeperForLedger(ledgerInfo);
+```
+
+Where:
+
+```java
+private BookKeeper resolveBookKeeperForLedger(LedgerInfo li) {
+    var resolver = config.getBookKeeperResolver();
+    return resolver != null ? resolver.apply(li) : this.bookKeeper;
+}
+```
+
+**Scope.** Pure additive; when resolvers are unset, behavior is identical.
+~80 lines of change in `managed-ledger/`, mostly mechanical substitutions.
+
+### S3. `SchemaStorageBookKeeperProvider` SPI
+
+**Problem.** `BookkeeperSchemaStorage` captures `this.bookKeeper` at
+`start()` time (line ~101). It directly opens schema ledgers via that single
+client. To support per-ledger routing we need to inject a resolver and to
+attribute each schema ledger to a cluster.
+
+**Hook (two parts).**
+
+1. **Add a tiny extension slot to `SchemaStorageFormat.proto`** — this is the
+   *only* proto change in this PIP:
+
+   ```protobuf
+   message PositionInfo {
+       required int64 ledgerId = 1;
+       required int64 entryId = 2;
+       repeated KeyValue properties = 3;   // ★ NEW — generic extension slot,
+                                           // mirrors PIP-404 LedgerInfo.properties
+   }
+
+   message KeyValue {
+       required string key = 1;
+       required string value = 2;
+   }
+   ```
+
+   This is the minimal possible change: a generic `properties` slot symmetric
+   to PIP-404. It does **not** mention bookie clusters; any future plugin can
+   use it.
+
+2. **Add a provider SPI for the schema storage BK client:**
+
+   ```java
+   // pulsar-broker/.../service/schema/SchemaStorageBookKeeperProvider.java
+   public interface SchemaStorageBookKeeperProvider {
+       /** Returns the BK client to use for the given schema position. */
+       BookKeeper resolve(PositionInfo position);
+       /** Returns the BK client to use for *creating* a new schema ledger. */
+       BookKeeper active();
+   }
+   ```
+
+   `BookkeeperSchemaStorage` resolves its provider as:
+
+   ```java
+   var provider = pulsar.getSchemaStorageBookKeeperProvider();   // nullable
+   BookKeeper bk = provider != null ? provider.resolve(position) : this.bookKeeper;
+   ```
+
+   The default (provider == null) preserves today's behavior exactly.
+
+**Scope.** One proto field (additive, `repeated` is forward/backward
+compatible in proto2) + one SPI interface + ~50 lines in
+`BookkeeperSchemaStorage`. No new REST endpoint, no CLI change.
+
+### Summary: Core Diff Footprint
+
+| Component | LOC | Risk |
+|-----------|-----|------|
+| `ManagedLedgerConfig` (S1, S2 setters/getters) | ~40 | Trivial additive |
+| `ManagedLedgerImpl` / `ManagedCursorImpl` resolver wiring | ~80 | Mechanical, well-scoped |
+| `SchemaStorageFormat.proto` `PositionInfo.properties` (S3) | 1 field | Additive, proto2-safe |
+| `SchemaStorageBookKeeperProvider` SPI + `BookkeeperSchemaStorage` wiring | ~50 | Additive |
+| **Total core diff** | **~170 LOC, 1 new optional proto field** | **Zero behavioral change when SPI unused** |
+
+Everything else lives in the plugin NAR.
+
+## Plugin Module: `pulsar-bookie-cluster-switching`
+
+A new top-level Maven module that builds a NAR (`pulsar-bookie-cluster-switching-<version>.nar`).
+
+### Module Layout
+
+```
+pulsar-bookie-cluster-switching/
+├─ src/main/java/.../
+│   ├─ storage/
+│   │   ├─ MultiClusterManagedLedgerStorage.java    (implements ManagedLedgerStorage)
+│   │   ├─ MultiClusterStorageClass.java            (implements BookkeeperManagedLedgerStorageClass)
+│   │   └─ RoutingManagedLedgerFactory.java
+│   ├─ schema/
+│   │   └─ RoutingSchemaBookKeeperProvider.java     (implements SchemaStorageBookKeeperProvider)
+│   ├─ registry/
+│   │   ├─ BookieClusterConfigManager.java
+│   │   └─ BookieClusterConfigWatcher.java
+│   ├─ orchestrator/
+│   │   ├─ BookieClusterSwitchOrchestrator.java
+│   │   ├─ LedgerCopyUtil.java
+│   │   └─ IdgenPrecheckService.java
+│   ├─ rest/
+│   │   └─ BookieClusterAdminServlet.java           (implements AdditionalServlet)
+│   └─ cli/
+│       └─ BookieClusterCommandFactory.java         (implements CustomCommandFactory)
+└─ src/main/resources/META-INF/services/
+    ├─ org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet
+    └─ org.apache.pulsar.admin.cli.extensions.CustomCommandFactory
+```
+
+### Wiring on Broker Start
+
+The plugin is activated by setting in `broker.conf`:
+
+```properties
+managedLedgerStorageClassName=org.apache.pulsar.ext.bookieswitch.MultiClusterManagedLedgerStorage
+additionalServlets=bookie-clusters-admin
+additionalServletDirectory=./plugins
+```
+
+Then `ManagedLedgerStorage.create()` (PIP-384 entry point) reflectively loads
+`MultiClusterManagedLedgerStorage`. Inside its `initialize()`:
+
+1. Reads its own configuration namespace (prefix `bookieClusterSwitch.*`).
+2. Reads `/admin/bookie-clusters/*` from Broker-ZK to discover registered
+   clusters and resolve the initial `activeClusterName` (tier-1 →
+   Broker-ZK `status=ACTIVE` cluster; tier-2 → optional
+   `bookieClusterSwitch.currentClusterName` hint; tier-3 → derive from
+   `bookkeeperMetadataServiceUri` and auto-register as `ACTIVE`).
+3. Builds one `BookkeeperManagedLedgerStorageClass` per registered cluster
+   (each owns its own `BookKeeper` client and `StatsProvider`).
+4. Constructs a `RoutingManagedLedgerFactory` that decorates a base
+   `ManagedLedgerFactoryImpl` and injects the per-ledger BK resolver via the
+   S1/S2 hooks above.
+5. Starts `BookieClusterConfigWatcher` to receive switch directives.
+6. (Origin-broker-only after a switch) starts the orchestrator.
+
+The plugin's `AdditionalServlet` instance is independently discovered by
+`AdditionalServlets.load(…)` in `WebService`; the plugin's
+`CustomCommandFactory` is independently discovered by `pulsar-admin`'s NAR
+class loader.
+
+## Detailed Design
+
+### Per-Ledger Attribution Encoding
+
+| Ledger type | proto container | Encoding | Key |
+|-------------|-----------------|----------|-----|
+| Topic Data | `ManagedLedgerInfo.LedgerInfo.properties` (PIP-404, tag 6) | `KeyValue` | `_pulsar.bookieClusterName` |
+| Cursor | `ManagedCursorInfo.cursorProperties` (existing, tag 8) | `StringProperty` | `_pulsar.bookieClusterName` |
+| Schema | `SchemaStorageFormat.PositionInfo.properties` (NEW, tag 3, S3) | `KeyValue` | `_pulsar.bookieClusterName` |
+
+The reserved key prefix `_pulsar.` is **owned by Pulsar core** to avoid
+collisions with plugin- or user-defined properties. Plugin code uses a
+constant:
+
+```java
+public static final String ATTR_KEY = "_pulsar.bookieClusterName";
+```
+
+**Compatibility.**
+
+- **proto2 forward-compat:** `repeated KeyValue properties` already exists in
+  `LedgerInfo` (PIP-404). For `PositionInfo` it's added with a new tag (3),
+  matching the symmetric pattern.
+- **Old broker reading new metadata:** ignores `properties` it doesn't
+  understand; preserves them via `UnknownFieldSet` on rewrite (already the
+  case for PIP-404).
+- **New broker reading old metadata (no attribution):** the plugin's resolver
+  falls back to the active cluster (which equals the original cluster for
+  pre-switch ledgers; correct).
+
+### Write-Path Stamping
+
+`RoutingManagedLedgerFactory` wraps the base factory. When `ManagedLedger`
+asks for a new ledger:
+
+```java
+// Pseudocode of the stamping path
+String activeCluster = multiClusterStorage.getActiveClusterName();
+BookKeeper bk = multiClusterStorage.getBookKeeper(activeCluster);    // via S1
+LedgerHandle lh = bk.asyncCreateLedger(...);
+
+// In the same metadata CAS that records the new LedgerInfo:
+LedgerInfo info = baseInfo.toBuilder()
+    .addProperties(KeyValue.newBuilder()
+        .setKey(ATTR_KEY)
+        .setValue(activeCluster)
+        .build())
+    .build();
+```
+
+The same pattern applies to:
+
+- `ManagedCursorImpl.doCreateNewMetadataLedger` → stamps
+  `cursorProperties[_pulsar.bookieClusterName] = activeCluster`.
+- `BookkeeperSchemaStorage.createLedger` (via S3 provider) → stamps
+  `PositionInfo.properties[_pulsar.bookieClusterName] = activeCluster`.
+
+The stamp is written in the **same Broker-ZK CAS** as the business field
+(`ledgerId`, `cursorsLedgerId`, `position`), guaranteeing atomicity
+(invariant **I1**).
+
+### Read-Path Routing
+
+```java
+// RoutingManagedLedgerFactory -- injected as ManagedLedgerConfig.bookKeeperResolver (S2)
+BookKeeper resolve(LedgerInfo li) {
+    String cluster = li.getPropertiesList().stream()
+        .filter(kv -> ATTR_KEY.equals(kv.getKey()))
+        .map(KeyValue::getValue)
+        .findFirst()
+        .orElseGet(multiClusterStorage::getActiveClusterName);   // legacy fallback
+    return multiClusterStorage.getBookKeeper(cluster);
+}
+```
+
+Identical pattern for cursors (consulting `cursorProperties`) and schema
+positions (consulting `PositionInfo.properties`).
+
+**Invariant I4** (entry-level consistency within a single read): the resolver
+is consulted exactly once at `LedgerHandle` open time; the resulting
+`LedgerHandle` is bound to one BK client for the entire read of that ledger.
+
+### Switch Phases (Operator-Driven State Machine)
+
+```
+NOT_REGISTERED
+    ↓
+[Operator: pulsar-admin bookie-clusters register --name newCluster --uri … --status STANDBY]
+    ↓
+REGISTERED (STANDBY)
+    ↓
+[Operator: pulsar-admin bookie-clusters switch --target newCluster]
+    ↓ (inline precheck: idgen-long advanced)
+SWITCH_TRIGGERED ← Coordinator broker writes /admin/bookie-clusters/switch-target
+    ↓
+LIVE_DUAL_READ  ← All brokers' watchers fire; new writes stamped newCluster;
+    ↓             reads of pre-stamped ledgers routed by attribution
+BUILD_PROMOTE_CLEANUP
+    ├── BUILD     ← Coordinator copies Schema/Cursor ledger bytes to newCluster
+    │             with the SAME ledgerId; entries copied with the SAME entryId
+    ├── PROMOTE   ← Schema: coordinator CAS-rewrites SchemaLocator
+    │             (changes only PositionInfo.properties[ATTR_KEY])
+    │             Cursor: coordinator forwards POST /internal/promote-cursor
+    │             to topic owner; owner CAS in ManagedCursorImpl lock
+    └── CLEANUP   ← After rollback window, delete old-cluster copies; promote
+                  Broker-ZK status: oldCluster→DEPRECATED, newCluster→ACTIVE
+    ↓
+DONE / DONE_WITH_FAILURES
+
+(Operator-triggered, within window):
+ROLLBACK ← Reverse-CAS attribution back to oldCluster; new-cluster
+          copies scheduled for delayed deletion.
+```
+
+All states above live **inside the plugin**. The orchestrator persists its
+progress under `/admin/bookie-clusters/orchestrator/<switchId>/...`.
+
+### Phase 1: Registration
+
+```bash
+pulsar-admin bookie-clusters register \
+    --name new-cluster \
+    --metadata-service-uri zk+null://new-zk:2181/ledgers-v2 \
+    --status STANDBY
+```
+
+`BookieClusterConfigManager` (plugin) validates:
+
+- `metadata-service-uri` does not collide with any already-registered cluster.
+- For co-located deployments, the chroot must differ from existing clusters.
+
+### Phase 2: Precheck (Invariant I11)
+
+```bash
+pulsar-admin bookie-clusters precheck --name new-cluster
+```
+
+`IdgenPrecheckService` (plugin) verifies the new cluster's `/ledgers/idgen-long`
+is advanced beyond `max(ledgerId)` of the source cluster. Otherwise
+`asyncCreateLedgerAdv(ledgerId, …)` during BUILD would collide with
+newly-allocated ledgerIds. `POST /switch` invokes precheck inline; failure
+returns HTTP 409 (bypassable via plugin config in lab/staging only).
+
+### Phase 3: Switch Trigger
+
+```bash
+pulsar-admin bookie-clusters switch --target new-cluster
+```
+
+The receiving broker (origin) writes:
+
+```
+/admin/bookie-clusters/switch-target
+  body: { "target": "new-cluster", "origin": "https://broker-1:8080",
+          "epoch": 7, "ts": 1716256800000 }
+```
+
+All brokers' `BookieClusterConfigWatcher` fire and execute, in order:
+
+1. `MultiClusterManagedLedgerStorage.switchActiveCluster(newCluster)` —
+   hot-swap `activeClusterName`; new BK client lazily created if not present.
+2. First switch: enable dual-read in the read router; subsequent switches:
+   refresh epoch.
+3. Emit metrics.
+4. **Origin broker only:** start the orchestrator (`BUILD → PROMOTE → CLEANUP`).
+   Non-origin brokers do not start an orchestrator. If the directive body
+   lacks `origin`, brokers fall back to leader election among themselves
+   (using the existing `LeaderElectionService`) to pick exactly one
+   orchestrator.
+
+### Phase 4: BUILD (Schema + Cursor Ledger Bytes)
+
+The orchestrator runs **two parallel stages** with bounded concurrency.
+
+#### Schema Ledger BUILD (centralized)
+
+```
+for each schemaId in /schemas/* (parallel, bounded):
+  locator = brokerZk.get("/schemas/<schemaId>")
+  for each indexEntry in locator.indexList:
+    ledgerId = indexEntry.position.ledgerId
+    cluster  = readClusterAttr(indexEntry.position) ?? activeClusterAtFirstWrite
+    if cluster == oldCluster:
+      LedgerCopyUtil.copyLedgerPreservingIds(
+          ledgerId, oldBk, newBk,
+          customMetadata,           // identical to source ledger
+          ensembleSize, qw, qa)
+        // Internally:
+        //   newBk.asyncCreateLedgerAdv(ledgerId, …)        ← SAME ledgerId
+        //   for entryId in [0, lastAddConfirmed]:
+        //     newLh.addEntry(entryId, srcEntry)            ← SAME entryId
+        //   newLh.close()
+```
+
+#### Cursor Ledger BUILD (centralized)
+
+Identical pattern over `/managed-ledgers/<topic>/<cursorName>` znodes, copying

Review Comment:
   Thanks, the explanation makes sense to me. I agree that preserving the same ledgerId can make the promotion/rollback state machine much simpler, especially if the goal is to keep existing positions unchanged and only switch the cluster attribution.
   
   I have two related questions that may be worth clarifying in the PIP.
   
   First, if the design relies on copying ledgers with the same ledgerId into the target BK cluster, do we need to explicitly define the safety invariant around ledgerId collision? Since different BK clusters normally have independent ledgerId allocation spaces, the source ledgerId may already exist in the target cluster, or may need to be reserved/imported in a way that prevents future allocation conflicts. It would be helpful to document whether the target BK cluster is expected to be empty/dedicated, or whether the migration flow will explicitly detect/reserve/import these ledgerIds safely.
   
   Second, I wonder whether preserving the same ledgerId is strictly required for all metadata ledgers, especially schema/cursor ledgers. For normal data ledgers, I understand the motivation to keep "(ledgerId, entryId)" stable. But for schema/cursor ledgers, the ledgerId seems more like an internal storage reference. If same-ledgerId import is hard to guarantee, could schema/cursor use a newly allocated target ledgerId and record the mapping in metadata instead?
   
   For example, schema could potentially be modeled as "schema version -> source position / target position", with historical versions backfilled and new versions dual-written during the migration window. Cursor ledgers may also be able to use an explicit migrated target ledger id, as long as recovery/delete/cleanup paths consistently resolve it. This would be more complex than same-ledgerId copy, so I’m not suggesting it must replace the current design. I just wonder whether it is worth documenting as an alternative/trade-off, since it may make the design more applicable to non-empty target BK clusters.
   
   I may be missing some constraints here, so I’d be happy to hear your thoughts.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["wolfstudy (via GitHub)" <gi...@apache.org>]

wolfstudy commented on PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#issuecomment-4504309644

   @lhotari Fixed, PTAL again, thanks


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [feat][pip] PIP-477: Dynamic Bookie Cluster Switching for Brokers [pulsar]

["wolfstudy (via GitHub)" <gi...@apache.org>]

wolfstudy commented on PR #25774:
URL: https://github.com/apache/pulsar/pull/25774#issuecomment-4466761361

   > Thanks @wolfstudy for the proposal.
   > 
   > **My main concern is architectural rather than technical: I'd like this to build on the storage-layer pluggability introduced by [PIP-384](https://github.com/apache/pulsar/blob/master/pip/pip-384.md) rather than adding this complexity to core Pulsar.**
   > 
   > PIP-384 deliberately decoupled `ManagedLedgerStorage` and introduced `ManagedLedgerStorageClass` so that the BookKeeper client is owned by a storage class, and a broker can hold multiple storage classes — each topic resolving to one of them via persistence policy. The framing PIP-384 establishes is "this topic lives on this storage backend." The core essence of PIP-477 — moving writes to a new BK cluster while preserving readability of historical data on the old one — fits naturally as an extension of that framing: **a topic can move between storage classes, with per-ledger attribution determining which storage class serves each historical read**.
   > 
   > Concretely, I think the path forward looks like this:
   > 
   > 1. **Implement PIP-477 first as a custom set of implementations of the existing interfaces, outside of core Pulsar.** A custom `ManagedLedgerStorage` can hold a `Map<clusterName, BookkeeperManagedLedgerStorageClass>` and expose the active one — this covers what `DynamicBookKeeperClientFactory` does. The orchestrator state machine, the Broker-ZK registry under `/admin/bookie-clusters/*`, the watcher, and the CLI/REST surface can live in the plugin module. The REST endpoints under `/admin/v2/bookie-clusters/*` can be exposed via [`AdditionalServlet`](https://github.com/apache/pulsar/blob/master/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServlet.java) without touching core REST classes.
   > 2. **The per-ledger attribution slot already exists in core.** [PIP-404](https://github.com/apache/pulsar/blob/master/pip/pip-404.md) added `repeated KeyValue properties = 6` to `LedgerInfo` and exposed `ManagedLedger#asyncAddLedgerProperty` / `asyncRemoveLedgerProperty` / `asyncGetLedgerProperty` precisely as a generic extension slot for plugins. The `★ new` `repeated KeyValue properties = 7` proposed in §"proto Changes" of PIP-477 is duplicating what's already there — `bookieClusterName` can be encoded as a property in the existing `LedgerInfo.properties` slot today, with no core proto change. `ManagedCursorInfo` similarly already has `cursorProperties` (a `repeated StringProperty` at tag 8) which can carry the same attribution.
   > 3. **The CLI surface is also already pluggable.** [PIP-201](https://github.com/apache/pulsar/blob/master/pip/pip-201.md) added the `CustomCommandFactory` / `CustomCommandGroup` / `CustomCommand` SPI (in `pulsar-client-tools-api`), bundled as a `.nar`, so `pulsar-admin bookie-clusters …` can be shipped entirely in the plugin with no `pulsar-admin` core change.
   > 4. **Where there are real SPI gaps, add narrow hooks — don't inline the feature.** A custom `ManagedLedgerFactory` returned by the storage class already gets you most of the way; the remainder should be small, focused additions, narrower than PIP-477 and useful to anyone writing a custom storage class.
   > 5. **If the plugin proves valuable to other users of Pulsar later, it can be moved into the Pulsar tree** as a separate module shipped alongside the broker, opt-in via configuration. The ~99% of users who never use this feature pay zero cost in core surface area, the orchestrator code can evolve on its own cadence.
   > 
   > This approach also provides a forcing function for a cleaner abstraction. PIP-384 currently assumes "one topic, one storage class for its lifetime." Generalizing to "one topic, one storage class at a time, with attribution-driven routing for previously-written ledgers" is a natural and reusable extension — useful well beyond the BK-cluster-switching scenario (e.g., per-topic storage-tier migration, gradual rollout of an alternative `ManagedLedgerStorage` implementation).
   > 
   > Would you be open to reframing the PIP along these lines?
   > 
   > Happy to discuss further on the dev list or here.
   
   Thanks for the thorough review. I agree to reframe PIP-470 along the PIP-384 extension axis:
   
   - **Drop the new proto fields**. Encode bookieClusterName into LedgerInfo.properties (already added by PIP-404) and ManagedCursorInfo.cursorProperties (already exists). The remaining open question is SchemaStorageFormat.PositionInfo, where no equivalent properties slot exists today — I'll either propose a tiny additive slot or fold attribution into SchemaLocator properties as a follow-up.
   
   - **Ship as a plugin NAR module** (pulsar-bookie-cluster-switching) that provides a custom ManagedLedgerStorage holding a Map<clusterName, BookkeeperManagedLedgerStorageClass>. Active cluster selection, ZK registry under /admin/bookie-clusters/*, watcher, and orchestrator (BUILD/PROMOTE/CLEANUP) all live in the plugin.
   
   - **REST surface** moves to AdditionalServlet; CLI moves to PIP-201 CustomCommandFactory NAR.
   
   - **Narrow core hooks** I still need (these would be useful to any custom storage class, not just BK switching):
   
      - ManagedLedgerConfig.activeBookKeeperSupplier — per-call active BK resolution on the write path.
      - BookkeeperSchemaStorage.schemaReadBookKeeperResolver — analogous read-path resolution for schema ledgers.
      - An owner-broker-callable hook for cursor CAS under the ManagedCursorImpl lock (or expose it as an internal endpoint via AdditionalServlet).
   - **Opt-in via configuration**; the ~99% of users that don't switch BK clusters pay zero cost in core.
   
   I'll update the PIP draft accordingly and post the revised version on the dev list.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org