PR: https://github.com/apache/pulsar/pull/26692 Rendered doc: https://github.com/merlimat/pulsar/blob/mmerli/pip-495/pip/pip-495.md
----- # PIP-495: Scalable Topic Geo-Replication *Sub-PIP of [PIP-460: Scalable Topics](pip-460.md)* > **In one paragraph:** replicated entries are forwarded **verbatim** at steady > state — never > decompressed or re-batched — and clusters **converge their layouts on both > axes** (segment > boundaries *and* entry-bucket boundaries) under a single rule, *the finer > layout wins*. Re-batching > happens only while a layout change propagates. Replicated subscriptions are > out of scope and will > get a dedicated PIP (see [Out of Scope](#out-of-scope)). ## Motivation [PIP-460](pip-460.md) and its sub-PIPs ([PIP-468](pip-468.md) controller, [PIP-483](pip-483.md) auto split/merge, [PIP-486](pip-486.md) key-shared consumption) define scalable topics **within a single cluster**: a `topic://` backed by a per-cluster, continuously-changing DAG of `segment://` range segments. None addresses **geo-replication**. Classic partitioned topics replicate cleanly because their structure is *static and cluster-consistent* — the partition count is topic metadata that every cluster holds identically, so `partition = hash(key) % numPartitions` is stable everywhere and a replicated entry lands, intact, in the same partition it would locally. A scalable topic deliberately makes its structure **per-cluster and dynamic**: at any instant two clusters may have different segment counts, segment boundaries and entry-bucket counts. That is exactly what breaks naive replication — a source batch has no guaranteed place to land as-is. This PIP defines how to replicate a scalable topic's message stream across clusters while (a) keeping the **steady-state replication path free of decompression and re-batching**, exactly like the classic replicator; (b) requiring **no cross-cluster coordination barrier**; (c) preserving the **key-shared guarantees classic partitioned topics already provide**; and (d) leaving the [PIP-486](pip-486.md) wire format and dispatch model **unchanged**. ## Background knowledge **Classic geo-replication.** A replicator on the source cluster reads committed entries from the local topic through a durable cursor and re-publishes them to each remote cluster as a producer. It re-publishes an entry **as-is** — it deserializes only the outer `MessageMetadata`, stamps `replicated_from` and a source-position property, and forwards the intact entry; it never cracks open a producer batch ([`GeoPersistentReplicator.java`](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java)). Loops are prevented via `replicated_from`; the destination deduplicates on the source position per replicator producer name; the cursor's start position follows `replicationStartAt`; ordering is per-producer. **Scalable topics (PIP-460 family).** A `topic://` is a per-cluster DAG of `segment://` range segments, recorded in the cluster's **local** metadata store and driven by a per-topic controller leader. Routing is **client-side**: a producer watches the layout and keeps one per-segment producer, routing a key to the active segment whose range contains the **high 16 bits** of the key's `Murmur3_32` hash ([`SegmentRouter`](../pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/SegmentRouter.java)); there is no broker-side component that receives a `topic://` publish and picks a segment. A split always cuts a segment at its **midpoint**; a merge joins two adjacent active segments. Every layout mutation creates the child segments *and the subscription cursors on them* before publishing the new layout, so a controller-registered subscription never misses messages across a split. [PIP-486](pip-486.md) adds **entry-bucketing**: within a segment, a key maps to a bucket by the **low 16 bits** of the same hash (`hashB`, an independent ring); bucket boundaries are equal-width for the segment's bucket count `N` and immutable for the segment's life (changing `N` is a same-range rollover). Producers batch per bucket, and each entry carries its **effective `hashB` range** (`entry_hash_min`/`entry_hash_max` — the smallest and largest `hashB` actually present, necessarily within one bucket) in cleartext outer metadata. The broker routes each **whole entry** to the one consumer owning the bucket containing `entry_hash_min` ([`PersistentEntryBucketDispatcherMultipleConsumers`](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentEntryBucketDispatcherMultipleConsumers.java)); there is **no fan-out and no consumer-side filtering** — an entry straddling two buckets would silently break key affinity. Under [PIP-483](pip-483.md), auto-rebucketing only ever *raises* `N`, always to a power of two; the consumer selector spreads `N` buckets over however many consumers are connected. **Key-shared guarantees, restated.** Two separable properties: **affinity** (at most one consumer handles a given key at a time) and **order** (a key's messages are delivered in the order they were published *by each origin*). Classic partitioned topics preserve both within a cluster's subscription — replicated messages included — because the stable partition assignment funnels all of a key's messages, local and replicated, into one partition log. They do **not** provide a single order *across* clusters: within a partition on cluster C, local and replicated messages interleave in C's arrival order, which differs from every other cluster's. ## Goals ### In Scope - Replicate a scalable topic's **message stream** across clusters with **no coordination barrier**. - **Verbatim forwarding at steady state**: once layouts have converged, every replicated entry is stored on the destination exactly as it was on the source — no decompression, no re-batching. - Preserve, per cluster, the key-shared guarantees classic partitioned topics already give: **affinity** (one consumer per key, replicated messages included) and **per-origin order** per key. - Leave the [PIP-486](pip-486.md) wire format and whole-entry dispatch **unchanged**. - Degrade gracefully: correctness must never depend on clusters agreeing on a layout. ### Out of Scope - **Replicated subscriptions.** Subscription positions on a scalable topic are per-segment and per-bucket, so replicated subscriptions will use a different mechanism from today's marker-based one; they will be specified in a dedicated PIP. - A single **cross-cluster** total order per key — not provided (matches classic; reconcile via payload timestamps/versions). ## High Level Design ### Re-route, not mirror Each cluster routes **all** arrivals — local producers *and* replicated entries from peers — into **its own** current layout. Segment DAG metadata is never replicated; each cluster owns its own. A replicated entry for key K lands in C's segment for K, next to C's local entries for K. This is what buys the guarantees: because every message for K (whatever its origin) funnels into C's one segment for K, that segment's dispatch sends K to **one** consumer (affinity), and the segment's log fixes a single delivery order in which each origin's messages appear in origin order. It is the exact analogue of how a classic partitioned topic funnels all origins into one partition — except the "partition" is C's dynamically-chosen segment. Because routing is client-side, the replicator **is a scalable-topic producer** toward the remote cluster (PIP-460 already states that replication uses "the same routing logic used by producers"). There is no receiving-side ingress; the only decision the destination makes is the one PIP-486 already makes for every entry — which consumer owns the bucket the entry was stamped with. ### Verbatim is the steady state; re-batching only during convergence An entry can be forwarded **as-is** when its destination is unambiguous without reading its payload: 1. **Segment fit:** an active destination segment's range *covers the whole range of the source segment* the entry was read from. Then every key in the entry belongs to that destination segment, whatever the entry contains. 2. **Bucket fit:** the entry's stamped `entry_hash_min`/`entry_hash_max` lies within **one** bucket of that destination segment. Then PIP-486 dispatch routes it whole, correctly. When both hold the replicator sends the raw entry to that segment, stamping `replicated_from` and the source position exactly as the classic replicator does. When either fails, the entry is **re-batched**: decompressed, regrouped by destination segment *and* bucket, and re-serialized as several single-bucket entries. Re-batching is always correct; it is only expensive. The whole design is then about making the two conditions hold for every entry at steady state. Both conditions ask the destination to be **at least as coarse** as the source — a *finer* destination is what breaks them — and replication runs in both directions, so: **converged layouts must be identical on both axes**, and while a layout change propagates, only the traffic flowing *toward* the finer cluster pays a re-batch. The cost lands on the **sender that has not yet applied the change** (the lagging, coarser cluster), not on the cluster that made it. ### Convergence: the finer layout wins Layouts converge under one rule, on both rings: **a boundary exists on every cluster if any cluster has it.** - **Segments.** A range split on any cluster gets split everywhere. A merge happens only when every cluster's local policy finds the range cold — i.e. it is cold *everywhere*. - **Entry-buckets.** A segment's bucket count is the maximum any cluster has (buckets are equal-width, so the same `N` means the same boundaries on every cluster — **aligned buckets**). PIP-483 never lowers `N` automatically, so this axis is monotonic in practice. This is the design's stance, not a compromise: scalable topics favor **higher parallelism** — a quieter cluster carries a few more segments and buckets than it would alone, in exchange for a replication path that never touches payloads. The alternatives — a destination that fans a straddling entry out to several consumers, or keeping `N` local and re-batching coarse→fine traffic forever — are rejected in [Alternatives](#alternatives). The converged shape is a **join**: the union of every cluster's boundaries and the maximum of their bucket counts. A join is commutative and idempotent, so it needs **no leader and no barrier** — each cluster keeps deciding its own splits locally, learns its peers' current layouts, and folds them into its own with ordinary split/rebucket operations, at its own pace. Divergence exists only in the window between one cluster changing its layout and every cluster having folded that change in (and drained the sealed predecessors). For the join to be *reachable* by every cluster, all clusters must cut the ring at the **same points**. Splits are midpoint splits and rollovers keep a segment's range, so two clusters that start from the same layout and only ever merge **siblings** (the two halves of one earlier split) always hold layouts that nest, and every target boundary is then the midpoint of some local segment. Replicated topics therefore restrict merges to sibling pairs. For the one case where layouts do *not* nest — clusters that first meet with independently evolved layouts (replication enabled on an existing topic, or a creation race) — the controller gains a single primitive, **split at an explicit boundary**, which brings a cluster onto the union of boundaries in one pass; from then on the steady-state discipline applies. ### PIP-486 is unchanged The wire format and dispatch are exactly PIP-486's: the destination reads the stamped effective range and routes the whole entry to the owner of its bucket. The only cross-cluster conventions are the fixed hash function (already version-pinned) and identical bucket boundaries on converged segments. What this PIP adds is *policy* for replicated topics (sibling-only merges, `N` from the join, caps agreed across clusters) and a replicator that keeps entries whole whenever the layouts allow. Because the stamp is the batch's **effective** range — its actual min/max `hashB`, a subset of its nominal bucket — a batch narrower than its source bucket still fits a finer destination bucket, so many entries stay verbatim even *during* a bucket-count transition; PIP-486's own [Geo-Replication Considerations](pip-486.md#pulsar-geo-replication-considerations) describes exactly this interplay. ## Detailed Design ### The replicator **Read side — a replication cursor per segment, exactly as today.** Replication keeps the classic push mechanism. When a segment topic is loaded on its owning broker and its namespace is replicated to other clusters, the broker opens a durable replication cursor on the segment (`pulsar.repl.<cluster>`, one per remote cluster), reads committed entries in order and republishes them to the remote — the same trigger, cursor naming, source-position stamping and message-TTL handling as [`GeoPersistentReplicator`](../pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java) on a partition, with no controller-managed subscription in the loop. Only what the replicator *publishes to* changes: the remote `topic://`, through the write side below, which breaks a batch up only when it does not fit. (For segment topics this replaces the classic replicator that the load-time replication check would otherwise start toward a same-named remote segment.) Because the topic load completes only after the replication check has opened the cursor, the cursor exists before any producer can attach — so a child segment created by a split replicates from its very first entry, with nothing for the controller to arrange. `replicationStartAt` (`Earliest`/`Latest`) applies, as for classic topics, when replication is enabled on a topic that already holds data; the cursor honors the topic's message TTL like the classic one, which bounds how long an unreachable remote pins sealed segments. Two scalable-topic specifics sit on top of the classic mechanism: - **Parent-drain gate.** A segment's replicator starts publishing only once the replication cursors (for the same remote) of all its *parents* in the DAG have drained — the parents are sealed, so the target is fixed — checked with the same per-segment backlog query the controller uses for its own drain checks. Without it, a child's new entries for a key could overtake the parent's older ones and break per-origin order across a source-side split, merge or rollover. The child's backlog while it waits is bounded by the parent's backlog at seal time, as for stream consumers. - **GC gating.** The controller prunes a sealed segment only when every subscription has drained it; that check gains a term for the segment's replication cursors, so a sealed segment with un-replicated backlog is never deleted — the same pinning a classic replication cursor exerts on ledgers, bounded by the message TTL as above. **Write side — a scalable-topic producer.** The replicator watches the remote layout and holds one per-segment producer to the remote, like any producer. For each entry read from its source segment `S`: - **Fast path (steady state).** Find the active remote segment `D` with `D.range ⊇ S.range`. If it exists and `[entry_hash_min, entry_hash_max]` lies within one of `D`'s buckets, send the raw entry to `D`'s producer with `replicated_from` and the source position stamped. No decompression, no reshape. - **Slow path (convergence window).** Otherwise decompress, regroup the messages by remote segment and bucket, re-serialize each group as a single-bucket entry (stamped with its own effective range) and publish it to the group's segment. This is the only path that reads payloads or per-message keys. Entries without a stamp (legacy segments from [PIP-475](pip-475.md) migration) always take it. Two kinds of entry never hit the slow path: **unkeyed** batches (their `hashB` is 0, so they fit bucket 0 of any segment, and unkeyed messages may go to any segment), and **e2e-encrypted** entries (batching is automatically disabled for e2e-encrypted scalable topics, so each entry is a single message routable by its cleartext outer key — see [End-to-end encryption](#end-to-end-encryption)). **Ordering.** The replicator publishes entries in cursor order and does not advance the cursor past an entry until every part of it is acknowledged. A key lives in one bucket, so all of a key's messages from one source entry stay in one part; parts of one entry share no keys. Together with per-producer ordering to each remote segment, each origin's order per key is preserved on the destination. **Dedup and loops.** The replicator's producer name is per **source segment** (`pulsar.repl.<sourceCluster>.<segmentId>`), so the source positions each destination segment sees from a given producer are monotonic and the classic source-position dedup applies unchanged on the fast path (one source entry → one destination entry). On the slow path one source entry becomes several parts, so the source-position property gains a **part index** and the destination compares `(ledgerId, entryId, part)` lexicographically. Only failed parts are re-sent; a part whose failure was an ambiguous timeout across a destination seal may be duplicated in the successor segment — the same window every scalable-topic producer has across a seal, confined here to the slow path. Loop prevention is `replicated_from`, on both paths. **Slow-path backpressure.** The slow path is the only place replication spends CPU on payloads, and its cost lands on the sender that has not yet applied a layout change. It peaks when the backlog of a sealed, old-layout segment is replayed — after an outage, every entry of it straddles the new boundaries — and many topics on a broker can be converging at once. It is therefore bounded by a **broker-wide budget shared by all replicated topics**, not a per-topic one: - **Only the slow path draws on it.** Each re-batch acquires the entry's bytes from a broker-level rate limiter (bytes re-batched per second) and one of a bounded number of concurrency slots (a decompressed batch is held in memory while it is regrouped). The fast path never touches either, so verbatim replication — every other topic's, and this topic's own on converged ranges — is unaffected by construction. - **Fair.** A segment replicator holds at most one slow-path operation in flight, and waiting replicators are served round-robin, so one large old-layout backlog cannot monopolize the budget while a small topic's convergence window starves. - **Throttling pauses reads, never drops.** A replicator that cannot acquire simply stops reading its cursor until it can — the same backpressure the classic replicator applies when its producer queue is full. The backlog grows on the source, pinning the sealed segment under GC gating and bounded by the message TTL, exactly as any replication backlog; the budget is never zero, so replication cannot stall outright. - **Parts keep the codec.** Regrouped parts are re-compressed with the source entry's codec by default — that re-compression is most of the slow path's CPU — with an option to send them uncompressed and trade bandwidth for CPU. - **Steady-state consumers are budgeted too.** Two cases never converge and use the slow path for as long as they last: unstamped entries from [PIP-475](pip-475.md) legacy segments until migration completes, and a peer stuck at a stale version or behind a cap mismatch. They draw on the same budget under the same fairness, so they cannot starve a genuine convergence window. ### Layout convergence protocol Segment layouts live in each cluster's **local** metadata store, and this design assumes nothing is shared between clusters — not even the configuration store. Convergence is a **controller-to-controller exchange of each cluster's current layout**, built so that lost, duplicated or reordered messages, broker crashes and controller failover can *delay* convergence but never prevent it or leave two clusters settled on different shapes. **Local decisions apply immediately.** A cluster's [PIP-483](pip-483.md) policy keeps working exactly as it does today: a split or rebucket-up it decides is applied to the local layout at once (within the effective caps below). There is no separate "desired" layout to maintain or persist — the layout itself is the state, and it is already persisted with a monotonic **epoch** that every mutation advances. The one thing the policy no longer does on its own is *coarsen*: a merge it wants becomes a **cold mark** on the boundary (soft state, re-derived from local load on every evaluation) and takes effect only under the merge rule below. **Exchange.** The controller leader pushes its **current state** — the layout (boundaries and bucket counts), its cold marks and its local caps — to every cluster it replicates to, through the admin client the broker already keeps per remote cluster, at an endpoint that redirects to that cluster's controller leader like every other scalable-topic admin call. The receiver keeps the latest state of each peer **in memory** (a push carries the sender's layout epoch, and a receiver never replaces a peer's state with one of a lower epoch), **folds the sender's layout into its own before answering** — a push is acknowledged only once the resulting layout mutations are committed to the receiver's local store — and its answer carries its own current state back, so one round trip both applies the sender's layout at the peer and brings the peer's layout to the sender. The sender records, per peer, the epoch of its own layout that the peer last acknowledged: that is what "every cluster has applied the change" means operationally, and it is what the status endpoint and the convergence-lag metric report. Pushes go out immediately on any local layout change, are retried with backoff until acknowledged (a peer that cannot apply — a transient failure, a cap it did not yet know about — fails the call and is simply retried), and repeat on a periodic **anti-entropy** timer regardless of change. Nothing is relayed: the replication set is a full mesh, so every pair exchanges directly, and each cluster needs to know only its own peers' state. Nothing about peers needs to be persisted either: a new leader after a failover pushes its state to every peer and has every peer's state back within one round trip; until then it folds in what peers send but defers its own splits, merges and rebuckets. Entries of clusters no longer in the topic's replication set are ignored. Layout changes are rare (split cadence), so the N-1 calls per change are cheap; a leader hosting many topics may batch several topics' states into one call per peer. **Join, on receipt.** On receiving a peer's state the controller folds it into the local layout at once: every peer boundary it lacks is added by splitting the segment containing it (a midpoint split in steady state, an explicit-boundary split otherwise); every range where the peer's bucket count is higher is raised by a rebucket rollover — each an ordinary CAS-guarded layout mutation under the normal cursor-before-metadata ordering. The join only ever *adds* parallelism, so the order in which peers' states arrive does not matter. Ranges where the local layout still lags a peer's are *diverged*: the local replicator's traffic toward that peer on those ranges is re-batched until the mutation lands and the sealed predecessors drain, while the peer's traffic toward the local, coarser layout still fits and stays verbatim. **Merge rule.** A boundary is removed — the two segments it separates are merged — only when the local policy marks it cold **and** every peer's latest state either marks it cold or no longer has it. Merging is the one operation that must know about peers, which is exactly why per-peer state exists at all. A merge that turns out premature (a peer went hot between its mark and the merge) is undone by that peer's next split, which the join propagates; PIP-483's cooldowns bound the churn. Lowering a bucket count is not automated today; a manual merge or rebucket-down on a replicated topic is treated as a cold mark, so it takes effect once every cluster has issued it. **Effective caps.** Each state carries the cluster's `maxSegments` and `maxEntryBucketsPerSegment`; the effective cap is the **minimum** over the cluster and its peers, and the local policy never splits or rebuckets beyond it — so a cluster never produces a layout a peer cannot follow. A cap mismatch is reported, not silently overridden. **Why every cluster ends on the same shape.** The join is a union of boundaries and a maximum of bucket counts — commutative, associative and idempotent — so whatever order peers' states arrive in, and however often a push is duplicated or lost and retried, every cluster's layout converges to the join of all clusters' layouts once every state has reached every peer; pushing the whole state (never a delta), retrying until acknowledged and repeating periodically guarantee that it does, as long as each replicating pair's connectivity is eventually restored (which replication itself needs anyway), whatever crashed in between. Folding in a state is deterministic and idempotent (adding a boundary that exists or raising `N` to its current value are no-ops), so a controller that fails mid-way resumes from the persisted layout and the next round of peer states. Merges are the only coarsening, and they need every cluster's agreement, so they never leave two clusters apart for longer than one exchange. Hence once local policies settle, every cluster's layout equals the same join; while they keep changing, replication stays correct throughout, only re-batching more. **Creation and enable.** Creating a scalable topic in a namespace replicated to other clusters creates it on every cluster with the **same initial layout**, through the same admin fan-out classic partitioned-topic creation uses; the first exchange then finds identical layouts and nothing to do. Nothing depends on that fan-out winning every race, though: if a cluster meets a peer whose layout does not nest with its own — a create-on-lookup that beat the fan-out, or replication enabled on a topic that already existed on several clusters — the first exchange simply yields a union with non-midpoint boundaries, and the explicit-boundary split brings each cluster onto it in one pass. ### End-to-end encryption An e2e-encrypted batch is opaque and indivisible — the broker cannot read keys or split it — so the slow path is impossible for it. [PIP-486](pip-486.md) already ships the mitigation: **the client SDK disables batching for every e2e-encrypted scalable topic**, unconditionally (replication may be enabled after a topic is produced to, so no producer can know at publish time whether its entries will ever be re-routed). Each e2e entry is one message, still stamped with its `hashB`, and the replicator places it by its cleartext outer key — segment by the key's routing hash, bucket by the stamp — so it is **always verbatim**, even during a layout transition. This PIP adds no rule; it relies on that one. ### Guarantees contract Under geo-replication a scalable topic provides, **per cluster**, the same key-shared guarantees as a classic partitioned topic: - **Affinity:** one consumer per key within a subscription, replicated messages included. - **Per-origin order:** on each cluster, a key's messages from a given origin are delivered in that origin's publish order; different origins interleave in the destination's arrival order. - **No cross-cluster order:** the interleaving differs between clusters; applications that need cross-cluster reconciliation use payload timestamps/versions, exactly as with classic topics today. ### Public-facing Changes - **Binary protocol:** none beyond [PIP-486](pip-486.md). The replication source-position message property gains an optional part index (slow path only). - **Metadata:** none. The exchanged state is the existing layout record (its epoch orders pushes), and peers' states are held in memory only; nothing is shared across clusters. - **Controller:** replicated topics restrict merges to sibling pairs; one new layout primitive, split at an explicit boundary (used only when a peer boundary is not a midpoint of a local segment); PIP-483 splits and rebuckets apply immediately as today, within the effective caps, while its merges become cold marks applied under the merge rule. - **Configuration:** `replicationStartAt` reused as-is. New: peer-exchange anti-entropy interval and push retry backoff; the broker-wide slow-path budget (bytes re-batched per second, maximum concurrent re-batches) and whether re-batched parts keep the source codec (default) or go uncompressed. - **Admin/CLI:** one broker-to-broker endpoint for the peer exchange (redirects to the controller leader; not meant for operators); per-topic replication status: for each remote cluster, converged vs. diverged ranges, convergence lag, age of its last received state, cap mismatches, replication backlog per segment. - **Metrics:** replicated entries and bytes by path (**verbatim** vs **re-batched**); **diverged ranges** gauge (local layout lags a peer's); **convergence lag** (peer change → folded in and drained); **peer exchange** successes/failures per peer; **slow-path budget** utilization and time throttled per segment; **sealed segments pinned** by replication backlog. ## Monitoring At steady state the **re-batched** rate is zero. A non-zero rate that tracks a layout change and then subsides is normal — it is the convergence window. One that does not subside means some cluster is not folding in its peers' layouts: check its **diverged ranges** gauge and **convergence lag** (a cluster at a stale version, a controller whose pushes to a peer keep failing, or a cap mismatch all show up here). **Sealed segments pinned** climbing during a remote outage is expected and bounded by the message TTL; climbing without an outage points at a stalled replicator. The **slow-path budget** sits at saturation, with throttled time climbing, while an old-layout backlog replays after an outage — that is it doing its job; at steady state it reads zero, and a persistent draw with no layout change in sight means un-migrated legacy segments or a peer that is not converging (the re-batched rate by topic says which). ## Security Considerations The only encryption interaction is PIP-486's e2e batching rule, relied on above. The fast path uses cleartext outer metadata only (the stamped `entry_hash_min`/`entry_hash_max` and, for single messages, the outer key); the slow path decompresses but never decrypts, and e2e entries never reach it. The peer exchange rides the admin channel and credentials brokers already use toward peer clusters (the path that fans out partitioned-topic creation today) and its endpoint requires the same broker-level role; since a peer's state can only add boundaries or raise bucket counts within the effective caps, a misbehaving peer can cost parallelism, never correctness. Bucket and segment assignment remain internal to a subscription within a cluster. ## Backward & Forward Compatibility - Scalable topics are Pulsar 5+ only; geo-replication of them is a new protocol path, so every cluster in a topic's replication set must run a version with this PIP. A replicator starts toward a remote only once that remote has answered the peer exchange; a remote that does not implement it is reported as unsupported, not written to. - No persisted-format change: the exchanged state is the existing layout record and peers' states are not persisted; the segment DAG format is unchanged and still not replicated. - Enabling replication on an existing topic is supported (see *Creation and enable*); disabling it stops the exchange and removes the replication cursors as for classic topics, and peers drop the cluster's entry once it leaves the replication set; the local layout keeps whatever shape it had and evolves locally from there. ## Alternatives - **Fan-out at the destination for straddling entries** (store whole, dispatch to every consumer whose bucket it touches, filter consumer-side). Would let bucket counts stay local, but requires the shared-entry dispatch and cross-consumer ack aggregation that PIP-486 deliberately avoided, and consumers today do not even know which buckets they own. Rejected — the destination must never see a straddling entry, which is what aligned buckets guarantee. - **Keep `N` local and re-batch coarse→fine traffic permanently.** Correct, and it spares the quieter cluster a few buckets — at the price of decompressing every batch flowing toward the busier cluster forever. Rejected: steady-state re-batching is precisely what this PIP exists to avoid. - **Always re-batch** (no convergence at all). Simplest, always correct, and what most cross-cluster mirroring tools do — but it puts decompression and re-serialization on every replicated entry. Rejected for the same reason. - **Embed: replicate each origin's segments verbatim and union them per cluster.** Avoids all re-batching and needs no convergence, but a consumer's bucket lives in *every* origin's segments (the bucket hash is independent of the segment hash), so read fan-in grows with the number of origins; it requires mirroring each origin's segment *topology*; and it yields only per-origin logs with no single per-key funnel. Rejected. Active/passive deployments need no special mode either: a passive cluster has no local load, so it never adds boundaries of its own and its layout simply converges to the active cluster's shape. - **Coordinated (barrier) layout changes.** Make layouts cluster-consistent via a coordinated split. Rejected: the split *transition* is the problem, not the decision — in-flight entries batched under the old layout must either be held behind a stop-the-world write barrier or re-batched into the new layout anyway. The join gives the same end state with no barrier and a bounded re-batch window. - **Leader-proposed target** (one cluster's controller decides the shared layout). Unnecessary: the join is commutative and idempotent, so every cluster computes the same target with no election and no single point of failure. - **No merges on replicated topics.** With "finest wins" and no merges, the state is just the local layout, the push is just the local layout, and there is no per-peer state at all — the simplest possible protocol, bounded by the caps. But cold ranges never coalesce, and even a manual merge would be undone by the next push from a peer that has not merged yet. Kept as the fallback if the per-peer entries prove objectionable; the current design pays one small in-memory entry per peer to keep PIP-483's lazy merges. - **A shared configuration-store record** (per-cluster layouts in one record every controller watches). Simplest to reason about, and where classic partitioned-topic metadata lives — but scalable-topic layouts live in each cluster's *local* store, and a configuration store shared across clusters is not something this design can assume. Rejected; the peer exchange needs nothing beyond the admin connectivity replication already requires, and its convergence does not depend on any shared system. - **A system topic, or replication-stream markers, carrying layout changes.** A replicated system topic per namespace, or markers in the replication stream, would deliver a change to every peer — but neither tells the sender whether a peer *applied* it, so an acknowledgement needs a reverse channel anyway; markers land in a segment log on the destination rather than at its controller; and a system topic adds real infrastructure (topics, cursors, retention, GC) for a message that fits one RPC. Rejected in favor of the direct call, whose acknowledgement *is* the confirmation that the peer applied the layout. - **Fixed cluster-agnostic bucket grid.** A static grid caps consumer count and forces fine bucketing everywhere. Rejected — aligned buckets come from the join and grow only where a cluster needs them. ## Links <!-- Updated afterwards --> * Related: [PIP-460](pip-460.md), [PIP-468](pip-468.md), [PIP-475](pip-475.md), [PIP-483](pip-483.md), [PIP-486](pip-486.md) * Mailing List discussion thread: * Mailing List voting thread: -- Matteo Merli <[email protected]>
