https://github.com/apache/pulsar/pull/25516
-----
# PIP-468: Scalable Topic Controller
*Sub-PIP of [PIP-460: Scalable Topics](pip-460.md)*
## Motivation
[PIP-460](pip-460.md) introduces scalable topics as a new topic type
in Pulsar, built on a DAG of range segments that can be split and
merged to scale independently of the initial configuration. PIP-460
defines the overall vision but defers the detailed design of each
component to dedicated sub-PIPs.
This PIP specifies the **Scalable Topic Controller** — the broker-side
component responsible for:
1. **Segment lifecycle management** — creating, terminating, and
deleting the underlying segment topics that back a scalable topic.
2. **Segment layout coordination** — executing split and merge
operations with correct ordering guarantees so that no messages are
lost and all subscription cursors exist before producers are
redirected.
3. **Consumer assignment** — coordinating which consumers receive
messages from which segments for a given subscription.
4. **Leader election** — ensuring exactly one broker acts as the
controller for each scalable topic, with automatic failover.
5. **DAG watch sessions** — pushing topology updates to connected
clients (producers and consumers) when the segment layout changes.
Without a well-defined controller, split and merge operations would be
unsafe: producers could write to segments that have no subscription
cursors, consumers could miss messages during topology changes, and
concurrent layout modifications could corrupt the segment DAG.
---
## Design
### Architecture Overview
The controller subsystem consists of four layers:
```
┌──────────────────────────────────────────────────────────┐
│ Admin API Layer │
│ ScalableTopics REST + Segments REST │
└──────────────┬───────────────────────┬───────────────────┘
│ │
┌──────────────▼───────────────────────▼───────────────────┐
│ ScalableTopicService │
│ Per-broker singleton; manages controllers │
└──────────────────────────┬───────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────┐
│ ScalableTopicController (per topic) │
│ Leader-elected; owns layout, split/merge, consumers │
│ │
│ ┌─────────────────┐ ┌───────────────────────────────┐ │
│ │ SegmentLayout │ │ SubscriptionCoordinator (×N) │ │
│ │ (immutable DAG) │ │ per-subscription assignments │ │
│ └─────────────────┘ └───────────────────────────────┘ │
└──────────────────────────┬───────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────┐
│ ScalableTopicResources │
│ Metadata store access (read/write/watch) │
└──────────────────────────────────────────────────────────┘
```
### Broker Roles and Request Routing
A scalable topic involves three distinct broker roles. Understanding
which broker handles which operation is critical to the design.
#### Any broker — DAG watch sessions
A **DAG watch session** (scalable topic lookup) can be served by **any
broker** in the cluster. The session reads segment metadata from the
shared metadata store and registers a watch for changes. No state is
held on the broker beyond the watch registration. This means producers
and consumers can connect to any broker for topic discovery — the same
way regular topic lookups work today.
When the metadata changes (due to a split or merge performed by the
controller leader), the metadata store notification fires on whichever
broker is serving the watch session, and that broker pushes the
updated DAG to the client.
#### Controller leader — layout mutations and consumer assignment
Each scalable topic has exactly one **controller leader** across the
cluster, elected via the metadata store. The controller leader is the
only broker that can:
- Execute **split** and **merge** operations (the multi-step protocols
described below).
- Accept **consumer registrations** and compute segment-to-consumer assignments.
- **Notify consumers** of assignment changes after topology updates.
Clients discover the controller leader's broker URL through the DAG
watch session response (`controller_broker_url` field). StreamConsumer
and CheckpointConsumer clients then connect directly to the controller
leader to register and receive assignments.
Admin API requests for split and merge are routed to the
`ScalableTopicService` on any broker, which obtains (or creates) a
local `ScalableTopicController` that participates in leader election.
If the local controller is the leader, it executes the operation
directly. If not, the request must be redirected to the leader (or the
controller must first win the election). The split and merge admin
endpoints do not require the request to land on the leader broker —
the `ScalableTopicService` handles this transparently.
#### Segment-owning brokers — produce and consume
Individual **segment topics** (`segment://`) are regular persistent
topics from the broker's perspective. They are distributed across the
cluster by the standard Pulsar **load manager** and namespace bundle
assignment — the same mechanism used for `persistent://` topics. Each
segment topic is owned by exactly one broker at any time, determined
by its namespace bundle hash.
The controller leader does not need to own any of the segment topics.
When the controller needs to operate on a segment (create, terminate,
delete, discover subscriptions), it uses the **Segments admin API**,
which routes the request to whichever broker currently owns that
segment's namespace bundle. This decoupling means the controller can
coordinate segments spread across many brokers without requiring
co-location.
```
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Broker A │ │ Broker B │ │ Broker C │
│ │ │ (controller │ │ │
│ segment-0 │ │ leader) │ │ segment-2 │
│ segment-1 │ │ │ │ │
│ │ │ DAG watch │ │ DAG watch │
│ DAG watch │ │ sessions │ │ sessions │
│ sessions │ │ │ │ │
└─────────────┘ └─────────────┘ └─────────────┘
│
Split/merge ops
use admin API to
reach segments on
Broker A and C
```
In this example, Broker B is the controller leader but owns no
segments. When it executes a split of segment-0, it calls the Segments
admin API which routes to Broker A (the owner of segment-0).
### Components
#### ScalableTopicService
A per-broker singleton, created and owned by `BrokerService`. It
manages the lifecycle of all `ScalableTopicController` instances on
the broker.
**Responsibilities:**
- **Controller management**: maintains a map of active controllers
keyed by topic name. Creates controllers on demand via
`getOrCreateController()` and releases them on topic unload or broker
shutdown.
- **Topic lifecycle**: handles `createScalableTopic()` and
`deleteScalableTopic()` admin operations, including creating/deleting
the underlying segment topics via the Segments admin API.
- **Delegation**: routes `splitSegment()` and `mergeSegments()`
requests to the appropriate controller.
- **Leader state monitoring**: listens for leader election state
changes and re-attempts election when the current leader is lost.
#### ScalableTopicController
A per-topic coordinator. Only one instance across the cluster is the
**leader** for a given scalable topic, ensured by leader election
through the metadata store. The leader is the only instance that
modifies the segment layout or assigns consumers.
**State:**
| Field | Type | Description |
|-------|------|-------------|
| `topicName` | `TopicName` | The `topic://` name of the scalable topic |
| `currentLayout` | `SegmentLayout` | The current immutable snapshot
of the segment DAG |
| `subscriptions` | `Map<String, SubscriptionCoordinator>` |
Per-subscription consumer coordinators |
| `leaderState` | `LeaderElectionState` | Current leader election state |
| `leaderElection` | `LeaderElection<String>` | Metadata store leader
election handle; value is the broker URL |
**Key operations:**
- `initialize()` — loads the current metadata from the store, then
attempts leader election. The elected leader stores its broker service
URL so that clients can discover and connect to it.
- `splitSegment(segmentId)` — splits an active segment at its midpoint
(see [Split Protocol](#split-protocol)).
- `mergeSegments(segmentId1, segmentId2)` — merges two adjacent active
segments (see [Merge Protocol](#merge-protocol)).
- `registerConsumer(subscription, consumer)` — registers a consumer
and returns its initial segment assignment.
- `unregisterConsumer(subscription, consumer)` — removes a consumer
and triggers rebalancing.
- `getLeaderBrokerUrl()` — returns the current leader's broker URL,
used by clients to connect to the controller.
#### SegmentLayout
An immutable, versioned snapshot of the segment DAG. All mutation
methods (`splitSegment`, `mergeSegments`, `pruneSegment`) return a
**new** `SegmentLayout` instance — the original is never modified.
This ensures safe concurrent reads without locking.
**Fields:**
| Field | Type | Description |
|-------|------|-------------|
| `epoch` | `long` | Monotonically increasing version; incremented on
every layout change |
| `nextSegmentId` | `long` | Counter for assigning IDs to new segments |
| `allSegments` | `Map<Long, SegmentInfo>` | Complete DAG: all
segments (active + sealed) |
| `activeSegments` | `Map<Long, SegmentInfo>` | Filtered view: only
`ACTIVE` segments |
**Key operations:**
- `findActiveSegment(hash)` — returns the active segment whose hash
range contains the given hash value. Used by producers for message
routing.
- `splitSegment(segmentId)` — validates the segment is active,
computes the midpoint of its hash range, creates two child
`SegmentInfo` records covering `[start, mid]` and `[mid+1, end]`,
seals the parent, and returns a new layout with incremented epoch.
- `mergeSegments(id1, id2)` — validates both segments are active and
adjacent (the end of one equals `start - 1` of the other), creates a
merged `SegmentInfo` covering the combined range, seals both parents,
and returns a new layout.
- `getLineage(segmentId)` — returns the full ancestor + descendant
chain for a segment, used for DAG traversal during catch-up reads.
- `toMetadata()` / `fromMetadata()` — converts to/from the persisted
`ScalableTopicMetadata` format.
#### SegmentInfo
A record representing a single segment in the DAG:
| Field | Type | Description |
|-------|------|-------------|
| `segmentId` | `long` | Unique, monotonically increasing identifier |
| `hashRange` | `HashRange` | The `[start, end]` inclusive hash range
this segment covers |
| `state` | `SegmentState` | `ACTIVE` or `SEALED` |
| `parentIds` | `List<Long>` | Parent segments (empty for root segments) |
| `childIds` | `List<Long>` | Child segments (empty for leaf/active segments) |
| `createdAtEpoch` | `long` | Layout epoch in which this segment was
created (`0` for the initial segments) |
| `sealedAtEpoch` | `long` | Layout epoch in which this segment was
sealed; only meaningful when `state == SEALED`. `0` when the segment
is `ACTIVE`; this sentinel is unambiguous because sealing always
happens as part of a split or merge, which increments the epoch — so a
segment can never be sealed at epoch `0`. |
Factory methods: `SegmentInfo.active(id, range, epoch)` and
`SegmentInfo.sealed(...)`.
#### SubscriptionCoordinator
Manages segment-to-consumer assignments for a single subscription
within a scalable topic.
**Assignment strategy:** round-robin. Active segments are sorted by
hash range start; consumers are sorted by name. Segments are
distributed evenly across consumers. When a consumer is added or
removed, or when the layout changes (split/merge), the coordinator
recomputes assignments and pushes updates to affected consumers.
**Key operations:**
- `addConsumer(consumer)` — adds a consumer, rebalances, and returns
the full assignment map.
- `removeConsumer(consumer)` — removes a consumer and redistributes
its segments.
- `onLayoutChange(newLayout)` — called after a split or merge;
recomputes all assignments against the new set of active segments.
#### ConsumerRegistration and ConsumerAssignment
`ConsumerRegistration` is a record identifying a connected consumer:
| Field | Type | Description |
|-------|------|-------------|
| `consumerId` | `long` | Protocol-level consumer ID |
| `consumerName` | `String` | Stable name chosen by the client |
| `cnx` | `TransportCnx` | The connection to push assignment updates |
`ConsumerAssignment` is the assignment sent to a consumer:
| Field | Type | Description |
|-------|------|-------------|
| `layoutEpoch` | `long` | The layout epoch this assignment is based on |
| `assignedSegments` | `List<AssignedSegment>` | The segments assigned
to this consumer |
Each `AssignedSegment` contains: `segmentId`, `hashRange`, and
`underlyingTopicName` (the `segment://` topic name the consumer should
connect to).
### Metadata Store Schema
Scalable topic metadata is stored in the metadata store under a
well-known path structure:
```
/topics/{tenant}/{namespace}/{encodedTopicName}
```
The value at this path is a JSON-serialized `ScalableTopicMetadata`:
```json
{
"epoch": 3,
"nextSegmentId": 5,
"segments": {
"0": { "segmentId": 0, "hashRange": {"start": 0, "end": 65535},
"state": "SEALED", "parentIds": [], "childIds": [1, 2],
"createdAtEpoch": 0, "sealedAtEpoch": 1 },
"1": { "segmentId": 1, "hashRange": {"start": 0, "end": 32767},
"state": "SEALED", "parentIds": [0], "childIds": [3, 4],
"createdAtEpoch": 1, "sealedAtEpoch": 3 },
"2": { "segmentId": 2, "hashRange": {"start": 32768, "end":
65535}, "state": "ACTIVE", "parentIds": [0], "childIds": [],
"createdAtEpoch": 1, "sealedAtEpoch": 0 },
"3": { "segmentId": 3, "hashRange": {"start": 0, "end": 16383},
"state": "ACTIVE", "parentIds": [1], "childIds": [], "createdAtEpoch":
3, "sealedAtEpoch": 0 },
"4": { "segmentId": 4, "hashRange": {"start": 16384, "end":
32767}, "state": "ACTIVE", "parentIds": [1], "childIds": [],
"createdAtEpoch": 3, "sealedAtEpoch": 0 }
},
"properties": {}
}
```
The controller leader lock is stored at:
```
/topics/{tenant}/{namespace}/{encodedTopicName}/controller
```
The value is the broker service URL of the elected leader.
### DAG Watch Sessions
As described in [Broker Roles](#broker-roles-and-request-routing), any
broker can serve a DAG watch session because the segment metadata
lives in the shared metadata store — no controller leader involvement
is required. The session is established via the binary protocol:
1. **Client sends `CommandScalableTopicLookup`** with a
client-assigned `sessionId` and the `topic://` topic name.
2. **Broker creates a `DagWatchSession`** that:
- Loads the current `ScalableTopicMetadata` from the store.
- Resolves which broker owns each active segment's `segment://`
topic (via topic lookup).
- Reads the controller broker URL from the leader lock path.
- Registers a metadata store notification listener for changes to
the topic's metadata path.
3. **Broker sends `CommandScalableTopicUpdate`** with the initial
`ScalableTopicDAG` containing the full segment DAG, broker addresses,
and controller URL.
4. **On metadata change**, the notification listener fires. The broker
reloads metadata, re-resolves broker addresses, and pushes an updated
`CommandScalableTopicUpdate` to the client.
5. **Client sends `CommandScalableTopicClose`** to tear down the session.
The `ScalableTopicDAG` protocol message contains:
| Field | Type | Description |
|-------|------|-------------|
| `epoch` | `uint64` | Layout epoch |
| `segments` | `repeated SegmentInfoProto` | Full segment DAG |
| `segment_brokers` | `repeated SegmentBrokerAddress` | Broker URL for
each active segment |
| `controller_broker_url` | `string` | Controller leader's broker URL |
| `controller_broker_url_tls` | `string` | TLS variant |
---
## Split Protocol
Splitting a segment is the core scaling-out operation. The protocol is
carefully ordered to guarantee that **no messages are lost** and **all
subscription cursors exist before producers are redirected**.
### Step-by-step
```
Controller Metadata Store Broker(s)
│ │ │
1. Discover │──── getSubscriptions ───────>│ │
subscriptions│<─── [sub-A, sub-B] ─────────│ │
│ │ │
2. Create │──── createSegment(child1, ──>│ ─── route ────────> │
children │ [sub-A, sub-B]) │ │
create topic
│──── createSegment(child2, ──>│ ─── route ────────> │ + cursors
│ [sub-A, sub-B]) │ │
│ │ │
3. Terminate │──── terminateSegment ───────>│ ─── route ────────> │
parent │ (parent) │ │ seal ML
│ │ │
4. Update │──── CAS update ─────────────>│ │
metadata │ (atomic) │ │
│ │ │
5. Notify │──── push to consumers ──────────────────────────> │
consumers │ (rebalance) │ │
```
**Step 1: Discover subscriptions.** The controller queries the parent
segment topic for its list of subscriptions. It first checks locally
(if the segment is on this broker), falling back to the admin API for
remote segments.
**Step 2: Create child segment topics.** Two new `segment://` topics
are created via the Segments admin API. Each creation request includes
the list of subscriptions discovered in step 1. The Segments API
handler creates the persistent topic and initializes subscription
cursors at the `Earliest` position. The admin API routes to whichever
broker owns the namespace bundle for each segment, so child segments
may be created on different brokers.
**Step 3: Terminate parent segment.** The parent segment topic is
terminated via the Segments admin API, which writes a termination
marker to the managed ledger. After termination, producers writing to
the parent receive `TopicTerminated` and must re-route to child
segments.
**Step 4: Atomic metadata update.** The controller issues a
compare-and-swap (CAS) update to the metadata store. The update
transitions the parent to `SEALED` state with child pointers, and adds
the two new `ACTIVE` child segments. The CAS guarantees atomicity — if
another operation modified the metadata concurrently, the update fails
and must be retried.
**Step 5: Notify consumers.** All `SubscriptionCoordinator` instances
are notified of the layout change. They recompute segment-to-consumer
assignments and push updates to connected consumers.
### Why this ordering matters
- Steps 1-2 **before** step 4: if we updated metadata first, clients
would discover the new segments immediately. Producers would start
writing to child segments that have no subscription cursors yet,
causing those messages to be missed by consumers.
- Step 3 **before** step 4: terminating the parent before the metadata
update ensures that by the time clients see the new layout, the parent
is already sealed. There is no window where both parent and children
are writable.
- Step 2 creates cursors at `Earliest`: this is safe because the child
topics are brand new (empty). The cursor starts at the beginning, and
the first message written by a redirected producer will be the first
message consumed.
---
## Merge Protocol
Merging two adjacent segments is the core scaling-in operation. The
protocol follows the same ordering principle: **create first, update
metadata last**.
### Step-by-step
**Step 1: Discover subscriptions.** The controller queries both parent
segments for their subscriptions and takes the union, so no
subscription is lost.
**Step 2: Create merged segment topic.** A single new `segment://`
topic is created with all discovered subscriptions.
**Step 3: Terminate both parent segments.** Both parents are
terminated via the admin API. This is done sequentially to avoid a
race where producers of one parent are still writing while the other
is already sealed.
**Step 4: Atomic metadata update.** The CAS update seals both parents
with child pointers to the merged segment, and adds the new `ACTIVE`
merged segment.
**Step 5: Notify consumers.** Same as split — rebalance and push.
### Cross-broker coordination
Unlike splits, merges inherently involve multiple brokers because the
two parent segments may be owned by different brokers (assigned by the
load manager). The controller leader handles this transparently: all
segment operations (create, terminate, delete) go through the Segments
admin API, which routes each request to the broker that currently owns
the target segment's namespace bundle. The controller does not need to
know or care which broker owns which segment — the admin API routing
handles it.
---
## Segment Topic Management via Admin API
Segment topics are managed exclusively through a dedicated REST API at
`/admin/v2/segments`. This is a deliberate design choice: segment
topics use the `segment://` domain and must not be confused with
regular `persistent://` topics.
### Endpoints
| Method | Path | Description |
|--------|------|-------------|
| `PUT` | `/{tenant}/{namespace}/{topic}/{descriptor}` | Create a
segment topic. Optional request body with subscription names to
pre-create. |
| `POST` | `/{tenant}/{namespace}/{topic}/{descriptor}/terminate` |
Terminate (seal) a segment topic. |
| `DELETE` | `/{tenant}/{namespace}/{topic}/{descriptor}` | Delete a
segment topic. |
The `{descriptor}` is the segment's hash range and ID in the format
`{hexStart}-{hexEnd}-{segmentId}` (e.g., `0000-7fff-1`).
### Create Segment
When creating a segment topic, the handler:
1. Constructs the full `segment://` topic name from the path components.
2. Validates namespace bundle ownership (the request is routed to the
owning broker).
3. Creates the persistent topic via `BrokerService.getOrCreateTopic()`.
4. For each subscription in the request body, creates a cursor at the
`Earliest` position.
### Terminate Segment
Terminates the managed ledger backing the segment topic. After
termination, the topic accepts no further writes, and producers
receive `TopicTerminated`.
### Delete Segment
Deletes the segment topic and its managed ledger. Used during scalable
topic deletion to clean up all underlying storage.
---
## Consumer Assignment
Consumer assignment behavior depends on the consumer type:
- **StreamConsumer** (ordered, cumulative ack) — requires coordinated
segment-to-consumer assignment from the controller leader. Each active
segment is owned by exactly one consumer at any time to preserve
per-segment ordering. This is the subject of the rest of this section.
- **CheckpointConsumer** (unmanaged, for connectors) — same as
StreamConsumer from a controller perspective: it registers with the
controller leader and receives an explicit segment assignment.
- **QueueConsumer** (unordered, individual ack) — **does not require
any controller-side assignment**. Because there is no ordering
requirement, every queue consumer attaches directly to **all**
segments of the scalable topic — both `ACTIVE` and `SEALED` — and each
segment-owning broker independently performs round-robin delivery
across the queue consumers attached to that segment. New segments
produced by a split are picked up transparently via the DAG watch
session push; sealed segments keep serving messages until their
backlog drains and are then dropped from the set. The controller
leader is not in the data path for queue consumers and no
`SubscriptionCoordinator` is involved for them.
### Registration Flow (StreamConsumer / CheckpointConsumer)
1. A `StreamConsumer` connects to the controller broker (discovered
via the DAG watch session's `controller_broker_url`).
2. The consumer sends a registration request with its `subscription`
name and `consumerName`.
3. The controller's `registerConsumer()` method routes to the
appropriate `SubscriptionCoordinator` (created on first use for that
subscription).
4. The coordinator adds the consumer, recomputes assignments, and
returns the consumer's `ConsumerAssignment` — a list of `(segmentId,
hashRange, segmentTopicName)` tuples.
5. The consumer connects to each assigned `segment://` topic and
begins consuming.
### Rebalancing
Rebalancing occurs when:
- A consumer is added or removed.
- A split or merge changes the set of active segments.
The rebalancing algorithm is round-robin:
1. Collect all active segments, sorted by hash range start.
2. Collect all consumers, sorted by name (for deterministic ordering).
3. Assign segments to consumers in round-robin order.
4. For each consumer whose assignment changed, push the new
`ConsumerAssignment`.
### Consumer Session Lifecycle
A consumer's registration and its segment assignment are treated as a
**persistent session**, not as a transient association tied to a TCP
connection. This is a deliberate departure from the existing Pulsar
consumer model, where consumer liveness is asserted purely by the TCP
connection to the broker.
**Rationale.** Scalable topic consumer assignments have non-trivial
cost: when a consumer disappears, its segments are redistributed among
the remaining consumers, each of which may need to reconnect to
different segment brokers and (for ordered consumers) drain in-flight
messages before the new assignment takes effect. Treating a brief
disconnection as a full consumer loss would cause unnecessary
rebalancing in common scenarios such as a consumer process restart or
a broker restart.
**What is persisted vs. in-memory.**
The **session itself is persisted** — the keep-alive state is not.
- **Persisted in the metadata store (the session):** the
`ConsumerRegistration` — consumer name and its current segment
assignment — keyed by `consumerName` under the subscription path. Once
a consumer successfully registers, this entry is durable and outlives
TCP disconnects, client restarts, and controller leader failovers. The
assignment survives failover without forcing consumers to re-register.
- **In-memory on the controller leader (the keep-alive):** whether
each consumer is currently connected, and the grace-period timer that
runs while it is disconnected. Keep-alive signals are **not** written
to the metadata store; the leader observes the consumer's connection
state directly and tracks the timer in RAM. This avoids a metadata
store write on every liveness tick.
**Session semantics (steady state).**
- When a consumer registers, the `SubscriptionCoordinator` writes its
`ConsumerRegistration` and marks it connected in-memory.
- If the consumer's connection drops, the leader does **not**
immediately remove the consumer. It transitions the in-memory state to
"disconnected" and starts a configurable **session grace period**
timer for that consumer.
- If the same `consumerName` reconnects within the grace period, the
timer is cancelled and the consumer resumes with the same persisted
assignment — no rebalance, no other consumer disturbed.
- If the grace period expires, the leader deletes the
`ConsumerRegistration` from the metadata store and triggers a
rebalance for the remaining consumers.
**Behavior on controller leader failover.**
Because liveness timers are in-memory only, they are lost when the
leader broker crashes. The new leader:
1. Loads all `ConsumerRegistration` entries for each subscription from
the metadata store, restoring the prior segment assignment.
2. Treats every consumer as "just disconnected" and starts a fresh
grace-period timer for each. No timestamps carry across from the
previous leader.
3. Consumers reconnecting to the new leader within the fresh grace
period resume with the same assignment — seamless from their
perspective.
4. Consumers that fail to reconnect before the fresh timer expires are
evicted and their segments are redistributed.
This means a leader failover always gives clients a full grace period
to reconnect, regardless of how long they had already been
disconnected under the old leader. The trade-off is explicit: we keep
the metadata store writes proportional to real membership changes
(register / unregister / rebalance) rather than to liveness ticks.
**Covered scenarios.**
- **Consumer process restart.** A client process restarts and
reconnects with the same `consumerName` well within the grace period —
the same segments are reassigned, no redistribution.
- **Controller leader restart.** The leader is re-elected on a
different broker. Consumer entries and assignments are restored from
the metadata store; the new leader starts a fresh grace-period timer
for every consumer so reconnects are transparent.
- **Segment-owning broker restart.** Segment topics are reassigned by
the standard Pulsar load manager; the consumer's segment-to-consumer
assignment is unaffected and reconnects to the new owning broker are
transparent at the consumer-assignment level.
The grace period and related tunables are configurable per broker and
will be specified in a follow-up sub-PIP.
### Layout Change Propagation
When a split or merge completes and the metadata is updated:
1. The controller reloads the metadata and creates a new `SegmentLayout`.
2. `notifySubscriptions()` is called, which iterates all
`SubscriptionCoordinator` instances.
3. Each coordinator calls `onLayoutChange(newLayout)`, which
recomputes assignments against the new active segments and pushes
updates to affected consumers.
---
## Leader Election and Failover
### Election
Each `ScalableTopicController` participates in leader election via the
metadata store's `LeaderElection` API. The election path is:
```
/topics/{tenant}/{namespace}/{encodedTopicName}/controller
```
The elected leader writes its broker service URL as the election
value. Clients discover the controller by reading this value
(delivered as part of the DAG watch session response).
### Failover
When the leader broker fails:
1. The metadata store detects the session loss and clears the leader lock.
2. Other brokers with active controllers for the same topic receive a
`NoLeader` state change notification.
3. The `ScalableTopicService.onLeaderStateChange()` handler re-invokes
`controller.initialize()`, which reloads metadata and re-attempts
election.
4. The new leader restores in-memory state from the persisted
metadata, including all subscriptions and their consumer registrations
with current assignments. Consumers reconnecting within the session
grace period find their sessions intact and resume with the same
segment assignment without a rebalance (see [Consumer Session
Lifecycle](#consumer-session-lifecycle)).
### Non-leader brokers
A broker that loses leadership or never wins it retains the controller
instance but all mutating operations (`splitSegment`, `mergeSegments`,
`registerConsumer`) throw `IllegalStateException` after a
`checkLeader()` guard. The controller remains available for read
operations like `getLayout()`.
---
## Scalable Topic Lifecycle
### Creation
```
Admin API (PUT /admin/v2/scalable/{tenant}/{namespace}/{topic})
└─> ScalableTopicService.createScalableTopic(topic, numInitialSegments)
├─> ScalableTopicController.createInitialMetadata(numInitialSegments)
│ └─> Divides [0x0000, 0xFFFF] into N equal ranges
│ Creates N SegmentInfo records (ACTIVE, no parents)
│ Returns ScalableTopicMetadata with epoch=0
├─> ScalableTopicResources.createScalableTopicAsync(topic, metadata)
│ └─> Writes metadata to store at /topics/{t}/{ns}/{topic}
└─> For each segment: createSegmentAsync via Segments admin API
└─> Creates segment:// topic on owning broker
```
### Deletion
```
Admin API (DELETE /admin/v2/scalable/{tenant}/{namespace}/{topic})
└─> ScalableTopicService.deleteScalableTopic(topic)
├─> releaseController(topic)
│ └─> Closes leader election, removes from controllers map
├─> Load metadata from store
├─> For each segment: deleteSegmentAsync via Segments admin API
│ └─> Deletes segment:// topic on owning broker (best-effort)
└─> ScalableTopicResources.deleteScalableTopicAsync(topic)
└─> Removes metadata from store
```
---
## Public-Facing Changes
### Binary Protocol
Three new protocol commands (added to `PulsarApi.proto`):
| Command | Direction | Field ID | Description |
|---------|-----------|----------|-------------|
| `CommandScalableTopicLookup` | Client → Broker | 70 | Initiates a
DAG watch session |
| `CommandScalableTopicUpdate` | Broker → Client | 71 | Initial
response and subsequent pushed updates |
| `CommandScalableTopicClose` | Client → Broker | 72 | Closes the DAG
watch session |
New protocol messages:
- `ScalableTopicDAG` — the full segment DAG with broker addresses and
controller URL.
- `SegmentInfoProto` — per-segment record in the DAG.
- `SegmentBrokerAddress` — maps a segment ID to its owning broker.
- `SegmentState` enum — `ACTIVE` or `SEALED`.
### Admin API
**Scalable Topics** (`/admin/v2/scalable`):
| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/{tenant}/{namespace}` | List scalable topics |
| `PUT` | `/{tenant}/{namespace}/{topic}` | Create scalable topic |
| `GET` | `/{tenant}/{namespace}/{topic}` | Get topic metadata |
| `DELETE` | `/{tenant}/{namespace}/{topic}` | Delete scalable topic |
| `GET` | `/{tenant}/{namespace}/{topic}/stats` | Get aggregated stats
for the scalable topic (segment counts, per-segment rates,
per-subscription state, consumer counts) |
| `PUT` | `/{tenant}/{namespace}/{topic}/subscriptions/{subscription}`
| Create a subscription on the scalable topic. The controller
propagates it to all active segments by issuing `createSubscription`
on each `segment://` topic via the Segments admin API. |
| `DELETE` | `/{tenant}/{namespace}/{topic}/subscriptions/{subscription}`
| Delete a subscription. The controller unregisters all consumers and
deletes the subscription from every segment. |
| `POST` | `/{tenant}/{namespace}/{topic}/split/{segmentId}` | Split a segment |
| `POST` | `/{tenant}/{namespace}/{topic}/merge/{segmentId1}/{segmentId2}`
| Merge two segments |
**Segments** (`/admin/v2/segments`):
| Method | Path | Description |
|--------|------|-------------|
| `PUT` | `/{tenant}/{namespace}/{topic}/{descriptor}` | Create segment topic |
| `POST` | `/{tenant}/{namespace}/{topic}/{descriptor}/terminate` |
Terminate segment |
| `DELETE` | `/{tenant}/{namespace}/{topic}/{descriptor}` | Delete
segment topic |
### Admin Client API
New `ScalableTopics` interface on `PulsarAdmin`:
- `listScalableTopics(namespace)` / `listScalableTopicsAsync(namespace)`
- `createScalableTopic(topic, numSegments)` / `createScalableTopicAsync(...)`
- `getMetadata(topic)` / `getMetadataAsync(topic)`
- `getStats(topic)` / `getStatsAsync(topic)`
- `deleteScalableTopic(topic)` / `deleteScalableTopicAsync(topic)`
- `createSubscription(topic, subscription)` / `createSubscriptionAsync(...)`
- `deleteSubscription(topic, subscription)` / `deleteSubscriptionAsync(...)`
- `splitSegment(topic, segmentId)` / `splitSegmentAsync(...)`
- `mergeSegments(topic, segmentId1, segmentId2)` / `mergeSegmentsAsync(...)`
- `createSegment(segmentTopic, subscriptions)` / `createSegmentAsync(...)`
- `terminateSegment(segmentTopic)` / `terminateSegmentAsync(...)`
- `deleteSegment(segmentTopic, force)` / `deleteSegmentAsync(...)`
### Configuration
| Property | Default | Description |
|----------|---------|-------------|
| `scalableTopicEnabled` | `true` | Enable scalable topic support on
the broker |
Additional configuration for auto-split thresholds and consumer
controller session grace period will be specified in follow-up
sub-PIPs.
### Metadata Store Paths
| Path | Content |
|------|---------|
| `/topics/{tenant}/{namespace}/{topic}` | `ScalableTopicMetadata`
JSON — segment DAG and global topic state |
| `/topics/{tenant}/{namespace}/{topic}/controller` | Leader broker
URL (ephemeral) |
| `/topics/{tenant}/{namespace}/{topic}/subscriptions/{subscription}`
| `SubscriptionMetadata` — subscription-level config and the persisted
set of consumer registrations |
|
`/topics/{tenant}/{namespace}/{topic}/subscriptions/{subscription}/consumers/{consumerName}`
| `ConsumerRegistration` — the durable session: consumer name and its
current segment assignment. Keep-alive state (connected/disconnected,
grace-period timer) is in-memory on the controller leader only and is
**not** persisted here. |
---
## Backward & Forward Compatibility
### Upgrade
The controller is a new component with no interaction with existing
partitioned or non-partitioned topics. Enabling `scalableTopicEnabled`
activates the new code paths. Existing topics are unaffected.
### Downgrade / Rollback
Scalable topic metadata and segment data remain in the metadata store
and BookKeeper. Rolling back to a version without scalable topic
support leaves this data intact but inaccessible. Upgrading again
restores access.
### Client Compatibility
The existing Pulsar client SDK rejects `topic://` and `segment://`
domains with a clear error message directing users to the V5 client
SDK. This check is enforced in `PulsarClientImpl` for producer,
consumer, and reader creation paths.
---
## Security Considerations
The controller follows the same tenant/namespace/topic authorization
model as existing Pulsar topics:
- Admin API operations require topic-level admin permissions.
- DAG watch sessions require topic-level lookup permissions.
- Consumer registration requires topic-level consume permissions.
- Segment operations are internal and authenticated via the broker's
internal admin client.
The controller leader lock in the metadata store is accessible only to
authenticated brokers.
---
## Links
- Parent PIP: [PIP-460: Scalable Topics](pip-460.md)
- V5 Client API: [PIP-466: New Java Client API (V5)](pip-466.md)
- Mailing List discussion thread: TBD
--
Matteo Merli
<[email protected]>
