You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/01 15:43:17 UTC
(pulsar) branch master updated: [doc][spec] PIP-494: Scalable Topics client specification (#26805)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new f61af25688f1 [doc][spec] PIP-494: Scalable Topics client specification (#26805)
f61af25688f1 is described below
commit f61af25688f1b601cef35ca0dedcef5d91e81dbd
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Oct 1 08:43:11 2026 -0700
[doc][spec] PIP-494: Scalable Topics client specification (#26805)
---
spec/scalable-topics/README.md | 78 +++++++
spec/scalable-topics/client-api.md | 189 +++++++++++++++
spec/scalable-topics/client-behavior.md | 141 ++++++++++++
spec/scalable-topics/compatibility.md | 70 ++++++
spec/scalable-topics/conformance.md | 70 ++++++
spec/scalable-topics/conventions.md | 80 +++++++
spec/scalable-topics/data-model.md | 107 +++++++++
spec/scalable-topics/error-model.md | 83 +++++++
spec/scalable-topics/java-mapping.md | 99 ++++++++
spec/scalable-topics/key-shared.md | 181 +++++++++++++++
spec/scalable-topics/stability.md | 130 +++++++++++
spec/scalable-topics/terminology.md | 167 ++++++++++++++
spec/scalable-topics/wire-protocol.md | 397 ++++++++++++++++++++++++++++++++
13 files changed, 1792 insertions(+)
diff --git a/spec/scalable-topics/README.md b/spec/scalable-topics/README.md
new file mode 100644
index 000000000000..3ff6fecf7d67
--- /dev/null
+++ b/spec/scalable-topics/README.md
@@ -0,0 +1,78 @@
+# Scalable Topics — Client Specification
+
+**Specification version:** 1.0
+**Status:** Stable. See [Stability and Versioning](stability.md).
+
+This is the authoritative, language-neutral specification for **Apache Pulsar Scalable Topics** clients.
+It defines what a conformant client SDK MUST do, in any language, to interoperate with an Apache Pulsar
+broker that implements scalable topics. The Java *V5 client* is the reference implementation; this
+specification — not any single implementation — is the source of truth.
+
+It is modeled on the conventions of authoritative cross-implementation specifications (e.g. the
+[OpenTelemetry Specification](https://opentelemetry.io/docs/specs/otel/)): formal normative language, a
+defined glossary, explicit stability and conformance rules, and a clear separation between the
+**API contract** (what an application observes) and the **implementation requirements** (what a client
+must do internally).
+
+## Audience
+
+- **SDK implementers** building a scalable-topics client in a new language — read all documents; the
+ [Conformance](conformance.md) document is the checklist.
+- **Application developers** — the [Client API](client-api.md) and [Data Model](data-model.md) documents
+ define observable behavior and guarantees.
+
+## Document set
+
+| # | Document | Status | Summary |
+|---|----------|--------|---------|
+| 1 | [Overview & Conventions](conventions.md) | Stable | Scope, normative language (BCP 14), document statuses, the language-neutral naming policy and Java mapping. |
+| 2 | [Terminology](terminology.md) | Stable | Normative definitions of every term used across the spec. |
+| 3 | [Stability and Versioning](stability.md) | Stable | Specification versioning, feature stability tiers, and backward/forward-compatibility guarantees. |
+| 4 | [Data Model](data-model.md) | Stable | Topic identity, the key/hash ring, message, message identifier, checkpoint, schema. |
+| 5 | [Client API](client-api.md) | Stable | The observable **contract**: producers, the three consumer modes, transactions — operations, inputs, guarantees, errors. Transport-agnostic. |
+| 6 | [Implementation Requirements](client-behavior.md) | Stable | What a conformant client MUST do internally: routing, layout tracking, per-segment fan-out, retries, multiplexing, concurrency, resource lifetime. |
+| 7 | [Wire Protocol](wire-protocol.md) | Stable | The complete Pulsar binary-protocol binding with sequence diagrams: command catalog, DAG-watch, consumer assignment, producing/consuming, namespace watch, transaction-coordinator discovery, auto-creation. |
+| 8 | [Error Model](error-model.md) | Stable | Error taxonomy, typed errors, retryability, and no-throw rules. |
+| 9 | [Conformance](conformance.md) | Stable | The binary conformance criteria and the per-feature requirements checklist. |
+| 10 | [Compatibility](compatibility.md) | Stable | Interoperability with classic (v4) topics, the migration bridge, and broker version requirements. |
+| 11 | [Scalable Key-Shared Consumption](key-shared.md) | Stable | Entry-bucketing ([PIP-486](../../pip/pip-486.md)): scaling ordered consumption beyond one consumer per segment, with per-key affinity. Normative; required on the consumer side for Stream consumers. |
+| A | [Java Reference Mapping](java-mapping.md) | Stable | Maps each language-neutral operation/type to the Java V5 client surface. |
+
+> **Status legend:** *Draft* = written, subject to change; *Stable* = ratified and frozen under the
+> versioning rules. Per-document and per-feature stability is governed by
+> [Stability and Versioning](stability.md).
+
+## Relationship to the design proposals (PIPs)
+
+Scalable topics are designed in the PIP series under [PIP-460](../../pip/pip-460.md): the controller
+([PIP-468](../../pip/pip-468.md)), auto split/merge ([PIP-483](../../pip/pip-483.md)), and key-shared
+consumption ([PIP-486](../../pip/pip-486.md)). The PIPs are **design rationale**; this specification is
+the **normative client contract**, covering behavior that is implemented or ratified via an accepted
+PIP. Scalable key-shared consumption ([PIP-486](../../pip/pip-486.md), document 11) is **normative**, not
+Experimental. Geo-replication remains out of scope until its design settles. A
+design that is documentable but still changing MAY appear marked **Experimental** (none currently). See
+[Stability](stability.md).
+
+## Change process
+
+This specification is **not itself a PIP**. It is a living normative document maintained under a
+dedicated change process:
+
+- **Every normative change to this specification MUST be made through a [PIP](../../pip/).** The PIP
+ carries the design and rationale for the change; reviewers evaluate the proposed behavior and the
+ spec edits together.
+- **The specification edits land *with* the PIP.** When a PIP that changes scalable-topic client
+ behavior is accepted, the corresponding edits to these documents are merged together with it —
+ typically in the **same pull request** — so the contract and its decision record never drift.
+- Accepted PIPs remain the **design rationale and decision record**; these documents are the
+ **normative contract**, kept continuously in sync with the ratified PIPs.
+
+The detailed mechanics (PIP-template fields for spec changes, and the handling of purely-editorial
+corrections) are for the project maintainers to define.
+
+## Publishing
+
+These documents are authored in the `apache/pulsar` repository and are intended to be published to the
+user-facing docs site, sourced from
+[`apache/pulsar-site`](https://github.com/apache/pulsar-site). The mechanism for mirroring them into
+`pulsar-site` is a separate task and is not defined here.
diff --git a/spec/scalable-topics/client-api.md b/spec/scalable-topics/client-api.md
new file mode 100644
index 000000000000..15365e729170
--- /dev/null
+++ b/spec/scalable-topics/client-api.md
@@ -0,0 +1,189 @@
+# Client API
+
+**Status:** Stable
+
+This document defines the **observable contract** of a scalable-topics client: the operations an
+application invokes, their inputs, and the guarantees they provide. It is transport-agnostic and
+language-neutral. *How* a client realizes these guarantees is in
+[Implementation Requirements](client-behavior.md); the wire encoding is in
+[Wire Protocol](wire-protocol.md); the Java surface is in [Java Reference Mapping](java-mapping.md).
+
+Terms in *italics* are defined in [Terminology](terminology.md). Operation names (e.g. *CreateProducer*,
+*Send*) are language-neutral; an SDK MUST provide each operation but SHOULD name and shape it idiomatically.
+
+## 1. Operation style
+
+- Every operation that may block on I/O MUST be available in a non-blocking (future/async) form. A
+ client SHOULD also provide a blocking form. The two forms MUST share the same semantics and resources.
+- This document states requirements once; they apply to both forms.
+
+## 2. Client and resource lifetime
+
+The **client** is the entry point and the factory for producers, consumers, and transactions.
+
+- A client is heavyweight and thread-safe. An application SHOULD create **one client instance and reuse
+ it for the application's lifetime**, sharing it across all producers and consumers. Creating a client
+ per operation is an anti-pattern.
+- A client MUST provide a **graceful close** (await pending operations) and MAY provide an immediate
+ shutdown (abandon pending operations). After close, the client MUST reject new operations.
+- Once *CreateProducer* / *Subscribe* / *Create* (consumer) succeeds, the returned producer or consumer
+ **MUST remain valid for its lifetime**. The client MUST recover transparently from transient failures
+ (disconnects, failovers, broker reassignment, layout changes). An application **MUST NOT** be required
+ to discard and recreate a producer or consumer in response to a publish or receive error; it retries
+ the *operation* on the same instance. A failed publish indicates that *message* failed, not that the
+ producer is broken.
+
+## 3. Producer
+
+A **producer** publishes messages to one scalable topic.
+
+### 3.1 CreateProducer
+
+**Inputs.** A topic name (REQUIRED) and optional configuration: producer name; *access mode* (default
+*shared*); send timeout; queue-full behavior; compression, batching, chunking, and encryption policies;
+initial sequence identifier; user properties.
+
+**Behavior & output.** On success, returns a ready producer. If the topic does not exist, the broker
+MAY *auto-create* it subject to policy (§7); otherwise the operation MUST fail with a *not-found* error.
+Under an *exclusive* access mode the operation MUST fail if exclusivity cannot be acquired (rather than
+deferring the failure to first publish).
+
+**Access modes.** A client MUST support: *shared* (many concurrent producers), *exclusive* (single
+writer), *exclusive-with-fencing* (single writer, fencing older ones), and *wait-for-exclusive* (block
+until exclusive can be acquired).
+
+### 3.2 Send
+
+A message is constructed and published with these settable attributes:
+
+| Attribute | Effect |
+|-----------|--------|
+| value | REQUIRED. Serialized with the producer's schema. |
+| key | Establishes *per-key ordering* and *routing*. Optional. |
+| properties | Application string key/values. |
+| event time | Application timestamp. |
+| sequence identifier | Overrides the auto-assigned deduplication id. |
+| delayed delivery | Make the message visible to consumers after a delay or at a time. |
+| transaction | Publish within a *transaction* (§6). |
+| replication clusters | Restrict geo-replication targets. |
+
+*Send* returns the published message's *message identifier*.
+
+**Guarantees.**
+
+- **Per-key ordering (REQUIRED).** All messages published with the same key MUST be delivered to an
+ ordered consumer in publish order, and this order MUST be preserved across segment *split/merge*.
+- **Keyless messages** carry **no** ordering guarantee.
+- **Ordering of concurrent async sends (REQUIRED).** For a single producer, messages MUST be published
+ in the order the application issued the *Send* calls (a later send MUST NOT overtake an earlier one
+ for the same key).
+- **Deduplication.** Each message carries a monotonically increasing *sequence identifier*. When topic
+ deduplication is enabled, the broker discards duplicates by (producer name, sequence identifier). A
+ client MUST expose the last successfully published sequence identifier.
+
+### 3.3 Other producer operations
+
+- **Flush** MUST await completion of all publishes issued before the flush call (publishes issued
+ afterward need not be awaited).
+- **Close** MUST complete pending publishes, release the producer's resources, and be idempotent.
+
+## 4. Consumers
+
+A client MUST provide three distinct consumer types. Ordering is always stated **per key** — *segments*
+are never an application concept (see [Terminology](terminology.md)).
+
+| Mode | Ordering | Acknowledgment | Position | v4 analog (informative) |
+|------|----------|----------------|----------|-------------------------|
+| Stream | per-key, ordered | cumulative | broker-managed cursor | Failover (ordered, load shared across the subscription) |
+| Queue | none (parallel, no key affinity) | individual + negative | broker-managed cursor | Shared |
+| Checkpoint | per-key, ordered | none (external position) | client-held *checkpoint* | reader-style |
+
+A client MUST provide a blocking single-message *Receive*, a *Receive* with timeout (returning nothing
+if it elapses), and SHOULD provide a bounded multi-message receive.
+
+### 4.1 Stream consumer (ordered, broker-managed)
+
+**Subscribe inputs.** Exactly one of a topic name **or** a namespace selector (§4.4) is REQUIRED, plus a
+subscription name (REQUIRED). Optional: initial position (applied only when the subscription is new),
+subscription properties, consumer name, acknowledgment-group time, read-compacted, replicate-subscription-state, encryption policy.
+
+**Guarantees.**
+
+- Messages MUST be delivered in **per-key order**.
+- Acknowledgment is **cumulative only**: acknowledging a message MUST mark every message up to and
+ including it (across the whole topic) as processed. A client MUST NOT offer individual ack in this mode.
+- Position is a **broker-managed cursor** keyed by subscription name; on reconnect the consumer MUST
+ resume from the persisted cursor, or from the configured initial position if the subscription is new.
+- Multiple consumers on one subscription **share the load** and rebalance as members join or leave,
+ while preserving per-key order (Failover-like).
+- Acknowledgment MAY be performed within a *transaction*, becoming effective on commit.
+
+### 4.2 Queue consumer (parallel, broker-managed)
+
+**Guarantees.**
+
+- Messages are distributed across the subscription's consumers for parallel processing with **no
+ ordering guarantee and no key affinity**: successive messages — even ones sharing a key — MAY go to
+ different consumers in any order. A queue consumer does not specify a key.
+- Acknowledgment is **individual** (each message separately); **negative acknowledgment** requests
+ redelivery of a single message. Position is a broker-managed cursor keyed by subscription name.
+- Acknowledgment MAY be performed within a *transaction*.
+- A client SHOULD support a **dead-letter** policy: messages exceeding a redelivery limit are routed to
+ a dead-letter topic.
+
+> *Informative:* per-key, single-consumer delivery is a property of the ordered modes only. To scale
+> ordered consumption beyond one consumer per segment — keeping per-key affinity — the *Stream*
+> consumer uses **entry-bucketing** ([Scalable Key-Shared Consumption](key-shared.md)), specified
+> separately; the application simply creates more stream consumers.
+
+### 4.3 Checkpoint consumer (unmanaged, external position)
+
+**Create inputs.** A topic name (REQUIRED); a start position as a *checkpoint* (default *latest*); an
+optional consumer-group name; consumer name; encryption policy. The terminal operation is *Create*
+(there is no broker subscription).
+
+**Guarantees.**
+
+- There is **no broker-managed cursor**. The application holds position as a *checkpoint*; the consumer
+ starts at the configured start position.
+- *Checkpoint* MUST capture a **consistent position across all of the topic's segments** and return an
+ opaque, serializable value. Restoring from a checkpoint MUST resume reading from exactly that position.
+- **Ungrouped** (default): the consumer independently reads the entire topic from its start position.
+- **Grouped**: consumers sharing a group name share the topic; each unit is read by exactly one member
+ and units rebalance as members join/leave. A group MUST NOT persist a cursor — each member resumes
+ from its own start position/restored checkpoint.
+
+### 4.4 Namespace subscription (stream)
+
+A stream consumer MAY subscribe to **all** scalable topics in a namespace, optionally filtered by topic
+properties (AND semantics; empty filter ⇒ all). Membership MUST be **live**: topics entering or leaving
+the matching set MUST cause the consumer to attach or detach automatically. Cumulative acknowledgment
+MUST behave as in §4.1, extended across the topic set. If the namespace watch cannot be established
+(e.g. namespace not found, authorization denied), *Subscribe* MUST fail (see [Wire Protocol](wire-protocol.md) §16).
+
+## 5. Errors
+
+Every operation reports failures as typed errors. Notably: a *not-found* error MUST be distinct (so an
+application can tell "topic absent" from a generic failure). The full taxonomy, retryability, and the
+rule that asynchronous APIs MUST NOT throw synchronously are in [Error Model](error-model.md).
+
+## 6. Transactions
+
+Transactions provide exactly-once semantics across topics and subscriptions: messages published and
+acknowledgments made within a transaction MUST become effective **atomically** on commit, or be
+discarded on abort or timeout.
+
+- **NewTransaction** returns a transaction handle with an open state and a timeout.
+- A message MAY be published within a transaction (§3.2); an acknowledgment MAY be made within a
+ transaction (§4.1, §4.2).
+- **Commit** MUST make all buffered publishes visible and all buffered acknowledgments durable, across
+ all involved topics. **Abort** MUST discard them.
+- A client MUST expose the transaction state and MUST fail commit/abort with a typed error when the
+ transaction is not in a committable/abortable state (already terminal, timed out, or errored).
+
+## 7. Auto-creation
+
+On connecting a producer or consumer to a non-existent scalable topic, the broker auto-creates it
+(with a single initial segment) **iff** auto-topic-creation policy permits, exactly as for classic
+topics. When policy forbids it, the connect MUST fail with a *not-found* error. A client MUST NOT assume
+a topic exists.
diff --git a/spec/scalable-topics/client-behavior.md b/spec/scalable-topics/client-behavior.md
new file mode 100644
index 000000000000..0d7b525f1990
--- /dev/null
+++ b/spec/scalable-topics/client-behavior.md
@@ -0,0 +1,141 @@
+# Implementation Requirements
+
+**Status:** Stable
+
+This document specifies what a conformant client MUST do **internally** to realize the
+[Client API](client-api.md) contract over the [Wire Protocol](wire-protocol.md). The API states the
+observable guarantees; this document states the mechanisms required to provide them. *Segments* are an
+internal concept (see [Terminology](terminology.md)) and appear here but never in the application-facing
+API.
+
+## 1. Layout tracking
+
+- A client MUST open a DAG-watch session ([Wire Protocol](wire-protocol.md) §4) for each topic it
+ produces to or consumes from with a queue consumer, and maintain a local copy of the current
+ `ScalableTopicDAG`.
+- A client MUST apply a pushed layout only if its `epoch` is greater than the last applied epoch.
+- A client MUST react to layout changes (split/merge, broker reassignment) **transparently** — opening
+ and closing per-segment producers/consumers as the active set changes — without surfacing the change
+ to the application.
+
+## 2. Producer routing
+
+For each message, a client MUST select the destination segment by this algorithm (so every client —
+including the v4 partitioned-topic compatibility path — routes a key identically):
+
+1. **Keyed:** `hash = raw_Murmur3_32(key_utf8) >>> 16` — the **high** 16 bits of the raw (unmasked)
+ 32-bit Murmur3 hash of the key's UTF-8 bytes; select the single `ACTIVE` segment whose
+ `[hash_start, hash_end]` contains `hash`. It is an internal error if no active segment covers the
+ hash (the active set MUST always tile the ring). The **low** 16 bits of the same hash are the key's
+ *entry-bucket* hash ([Key-Shared](key-shared.md) §2); a client computes the hash once per key and
+ splits it.
+2. **Keyless:** round-robin across the active segments.
+3. **Migration (all-legacy layout):** when every active segment is legacy, route a keyed message by
+ `signSafeMod(Murmur3_32(key_utf8), N)` over the `N` legacy segments (v4 partitioned-topic routing).
+
+A client MUST maintain at most one underlying producer per active segment, created **lazily** on first
+send to that segment. Under an *exclusive* access mode, a client MUST eagerly attach to every active
+segment at create time, so a collision surfaces from *CreateProducer* rather than first send.
+
+## 3. Ordering guarantees (producer side)
+
+- A client MUST publish all messages for one key in *Send*-call order, and MUST preserve that order
+ across a split/merge: when the target segment seals, the key's writes move to the active successor
+ without reordering.
+- For asynchronous sends, a client MUST preserve per-segment dispatch order = caller call order — a
+ later message MUST NOT enter a segment's send queue before an earlier one.
+
+## 4. Transparent segment-gone retry
+
+When the target segment is sealed or terminated between routing and send (a split/merge, or a migration
+cutover), a client MUST NOT fail the send. It MUST:
+
+1. drop the stale per-segment producer,
+2. wait (bounded backoff) for the DAG watch to deliver the new layout,
+3. re-route and retry,
+
+up to a bounded number of attempts (the reference client uses 10). Only on exhausting the attempts MUST
+the send fail. This retry is invisible to the application.
+
+## 5. Consumer fan-out and multiplexing
+
+A client MUST present each consumer as a single logical stream while reading from multiple segments:
+
+- **Queue consumer:** subscribe (Shared) to **all** active and sealed segments of the topic; multiplex
+ deliveries into one receive queue; route each individual ack/nack to the correct per-segment consumer
+ using the segment id carried in the message identifier; reconcile the per-segment subscription set
+ toward the latest layout as it changes.
+- **Stream consumer:** maintain one per-segment subscription for each **assigned** segment
+ ([Wire Protocol](wire-protocol.md) §6.2) — Exclusive when the consumer is the segment's sole owner,
+ `Key_Shared`-by-bucket when the assignment shares the segment ([Key-Shared](key-shared.md) §4);
+ multiplex into one receive queue.
+- **Checkpoint consumer:** maintain one per-segment Reader for each segment it reads (all segments when
+ ungrouped; assigned segments when grouped); track read positions client-side.
+
+## 6. Stream cumulative ack (position vector)
+
+The stream consumer's cumulative ack MUST advance **all** segments, not just the one the acked message
+came from. A client MUST:
+
+- record, for each delivered message, a **position vector** — a snapshot of the latest delivered message
+ identifier of every segment at the moment the message was enqueued; and
+- on a cumulative ack, cumulatively acknowledge every segment up to the positions recorded in that
+ message's vector. On a segment attached `Key_Shared`-by-bucket (a shared segment,
+ [Key-Shared](key-shared.md) §4) cumulative acknowledgment is forbidden by the broker: the client MUST
+ translate that segment's slice into **individual** acknowledgments of every delivered-but-unacked
+ message at or before the recorded position.
+
+For a **namespace** stream consumer the vector spans topics and segments (`{topic → {segmentId →
+messageId}}`), and the ack fans out to every per-topic consumer accordingly.
+
+Acknowledgment is **always explicit**. Closing the consumer, and a topic leaving the matching set, MUST
+NOT acknowledge anything on the application's behalf: a removed topic's per-topic consumer is simply
+detached, and any message that was delivered but not yet acknowledged is redelivered if the topic is
+later re-added (**at-least-once**). If a cumulative ack's vector references a topic that has since left
+the set, that topic's slice is skipped.
+
+## 7. Checkpoints
+
+A *checkpoint* MUST capture an atomic snapshot of the per-segment read positions of the consumer's
+segment set, serialized to an opaque value. Restoring MUST seek each segment's reader to its saved
+position. Checkpoint capture and restore involve no broker round-trip and no acknowledgment.
+
+## 8. Assignment session (ordered / grouped consumers)
+
+A client MUST:
+
+- discover the controller leader from the layout (`controller_broker_url`) and connect to it directly
+ (scalable-topic URIs are not resolvable via standard lookup);
+- register with `CommandScalableTopicSubscribe`, attach to exactly the assigned segments — in the mode
+ each segment's `bucket_ranges` selects ([Key-Shared](key-shared.md) §4) — and on each
+ `CommandScalableTopicAssignmentUpdate` apply the diff (attach added, detach removed, re-subscribe on
+ a mode flip) only if its `layout_epoch` is ≥ the last applied; applying MUST be idempotent for
+ unchanged segments;
+- before detaching a segment (dropped from the assignment, or re-subscribed on a mode flip), **drain**
+ it: stop delivering from that segment and wait until every already-delivered message of the segment is
+ acknowledged, so the close cannot discard or reorder in-flight messages ([Key-Shared](key-shared.md)
+ §4);
+- on connection loss, re-run lookup → connect → register → subscribe with backoff; within the
+ controller grace period the same assignment returns unchanged, past it the client applies the
+ rebalanced diff;
+- on close, send `CommandScalableTopicUnsubscribe` (best-effort — never fail the close on it) so the
+ controller unregisters and rebalances immediately; without it the registration lingers for the
+ grace period, or indefinitely while the client process lives ([Wire Protocol](wire-protocol.md)
+ §6.2, *Clean leave*).
+
+## 9. Resilience and resource lifetime
+
+- A producer/consumer MUST survive broker disconnects, failovers, reassignment, and layout changes by
+ reconnecting and re-syncing internally; it MUST remain valid (the application is not required to
+ recreate it — see [Client API](client-api.md) §2).
+- A client SHOULD coalesce concurrent reconcile/reconnect attempts so a single layout or connection
+ event does not trigger redundant work.
+
+## 10. Concurrency and thread-safety
+
+- Producers, consumers, and the client MUST be safe for concurrent use from multiple application threads.
+- A client MUST NOT block a network/IO thread on an operation that depends on that same thread's
+ response (e.g. creating a per-segment producer whose lookup completes on the IO thread); such paths
+ MUST be chained asynchronously.
+- Buffers and per-segment resources MUST be released on close; close MUST be idempotent and MUST complete
+ in-flight operations (graceful) per [Client API](client-api.md) §2.
diff --git a/spec/scalable-topics/compatibility.md b/spec/scalable-topics/compatibility.md
new file mode 100644
index 000000000000..42076486cee4
--- /dev/null
+++ b/spec/scalable-topics/compatibility.md
@@ -0,0 +1,70 @@
+# Compatibility
+
+**Status:** Stable
+
+This document specifies interoperability between scalable topics and classic (v4) Pulsar: the
+migration bridge from existing topics, broker version/feature gating, and forward/backward compatibility
+of the protocol additions.
+
+## 1. Migration bridge: addressing a classic topic as a scalable topic
+
+To let an existing deployment adopt scalable topics without rewriting every topic reference, a client
+MAY address an existing classic topic through the scalable-topic surface. The broker bridges it with a
+**synthetic layout**:
+
+- A name given as `persistent://tenant/ns/name` (or a short form normalized to
+ `persistent://public/default/name`) is resolved ([Wire Protocol](wire-protocol.md) §4) to a
+ **synthetic single-segment** `ScalableTopicDAG`. The single `SegmentInfoProto` carries
+ `legacy_topic_name` set to the wrapped `persistent://` topic and has **no** `segment://` name.
+- A client targets the wrapped `persistent://` name directly for that segment (producing, subscribing,
+ reading).
+- For a **partitioned** classic topic, the synthetic layout presents the partitions as legacy segments;
+ a keyed message routes by `signSafeMod(Murmur3_32(key), N)` over the `N` partitions — **identical to
+ v4 partitioned-topic routing** — so a topic keeps its existing key→partition mapping while migrated.
+
+This bridge exists only for migration. New applications use canonical `topic://` names
+([Data Model](data-model.md) §1).
+
+> *Informative:* a V5-managed connection to a wrapped `persistent://` topic may be marked so the broker's
+> migration pre-checks can distinguish it from an ordinary v4 connection. This is a broker-internal
+> detail and not part of the client contract.
+
+## 2. Coexistence with classic topics and clients
+
+- A scalable topic (`topic://`) and a classic topic (`persistent://`) are distinct identities; the
+ reused per-segment subscriptions and producers behave exactly as classic v4 ones at the segment level
+ ([Wire Protocol](wire-protocol.md) §10).
+- A v4 client cannot consume a `topic://` scalable topic directly (it has no DAG-watch/assignment
+ logic); it interoperates only through the migration bridge against the underlying `persistent://`
+ topic during migration.
+- Classic-topic clients and scalable-topic clients MAY operate on the same cluster simultaneously.
+
+## 3. Broker version and feature gating
+
+- `segment://` topics are served **only** by brokers that implement scalable topics. The
+ scalable-topic command family ([Wire Protocol](wire-protocol.md) §1) MUST NOT be sent to a broker that
+ does not advertise support; a client MUST negotiate support before using it.
+- **Transaction-coordinator discovery** via the assignment watch is gated by the
+ `supports_tc_metadata_discovery` feature flag. A client MUST use the TC discovery watch
+ ([Wire Protocol](wire-protocol.md) §8) only against a broker that advertises it, and fall back to the
+ standard coordinator-lookup path otherwise.
+- **Entry-bucketing** ([Scalable Key-Shared Consumption](key-shared.md)): a client MUST gate producer
+ stamping/bucketing on observed broker support and remain interoperable with brokers that lack it; the
+ consumer-side handling rides the assignment protocol and needs no separate gate (a broker without the
+ feature never sends `bucket_ranges`).
+
+## 4. Protocol forward/backward compatibility
+
+- The scalable-topic commands occupy dedicated `BaseCommand` field numbers and are additive; a broker
+ or client that does not understand them ignores them. They are never sent on a path where the peer is
+ not known to support the feature (§3).
+- The entry-bucketing `MessageMetadata` fields are **optional** and inert on classic topics; a broker
+ without the feature ignores them, and a producer without the feature simply omits them.
+- Serialized **message identifiers** and **checkpoints** MUST remain restorable across version upgrades
+ ([Data Model](data-model.md) §4–§5), because applications persist them externally.
+
+## 5. Schema and policies
+
+Schema negotiation, topic-level policies, authentication/authorization, and TLS are the standard Pulsar
+mechanisms, applied at the segment level. A scalable topic has a single topic-level schema regardless of
+segment count ([Data Model](data-model.md) §6).
diff --git a/spec/scalable-topics/conformance.md b/spec/scalable-topics/conformance.md
new file mode 100644
index 000000000000..6579241467f0
--- /dev/null
+++ b/spec/scalable-topics/conformance.md
@@ -0,0 +1,70 @@
+# Conformance
+
+**Status:** Stable
+
+This document defines what it means for a client to **conform** to this specification, separates the
+**required core** from **optional capabilities**, and gives a per-requirement checklist.
+
+## 1. Definition
+
+A **conformant client** implements the **required core** (§2) such that it interoperates with any
+conformant Apache Pulsar broker and with conformant clients of other languages operating on the same
+topics. A client MAY additionally implement any **optional capability** (§3); if it does, it MUST
+implement that capability per its specification (partial implementation of a capability is
+non-conformant).
+
+The single hardest interoperability requirement is **identical routing**: a conformant client MUST
+implement the producer routing algorithm ([Implementation Requirements](client-behavior.md) §2,
+[Wire Protocol](wire-protocol.md) §5) exactly, so that any client routes a given key to the same segment
+as any other.
+
+## 2. Required core
+
+A conformant client MUST provide all of:
+
+| # | Requirement | Spec |
+|---|-------------|------|
+| C1 | Resolve a scalable topic and track its layout via the DAG-watch session, applying updates by epoch monotonicity. | [Wire](wire-protocol.md) §4, [Impl](client-behavior.md) §1 |
+| C2 | Producer routing: keyed `Murmur3_32 & 0xFFFF` → covering active segment; keyless round-robin; all-legacy mod-N. | [Impl](client-behavior.md) §2 |
+| C3 | Per-key publish ordering, preserved across split/merge; async per-segment call-order. | [API](client-api.md) §3.2, [Impl](client-behavior.md) §3 |
+| C4 | Transparent segment-gone retry on producing. | [Impl](client-behavior.md) §4 |
+| C5 | Deduplication via monotonic sequence identifiers; expose last published id. | [API](client-api.md) §3.2 |
+| C6 | Producer access modes (shared, exclusive, exclusive-with-fencing, wait-for-exclusive) with eager exclusive attach. | [API](client-api.md) §3.1, [Impl](client-behavior.md) §2 |
+| C7 | Stream consumer: per-key ordered delivery; **cumulative** ack advancing all segments via position vector; broker-managed cursor; controller assignment + rebalance + reconnect/grace. | [API](client-api.md) §4.1, [Impl](client-behavior.md) §6, §8 |
+| C8 | Queue consumer: parallel, no-affinity delivery; **individual** ack + negative ack; attach to all active+sealed segments; per-segment ack routing. | [API](client-api.md) §4.2, [Impl](client-behavior.md) §5 |
+| C9 | Checkpoint consumer: external position; cross-segment atomic checkpoint capture/restore; ungrouped and grouped modes. | [API](client-api.md) §4.3, [Impl](client-behavior.md) §7 |
+| C10 | Opaque, serializable, comparable message identifiers and opaque, serializable checkpoints with cross-version restore. | [Data Model](data-model.md) §4–§5 |
+| C11 | Topic-level schema; broker-enforced compatibility surfaced as a typed error. | [Data Model](data-model.md) §6 |
+| C12 | Auto-creation honored; not-found surfaced as a distinct error category. | [API](client-api.md) §7, [Errors](error-model.md) §2 |
+| C13 | Error model: typed categories, async-never-throws-synchronously, internal-retry classification. | [Errors](error-model.md) |
+| C14 | Resource lifetime: long-lived, self-healing producers/consumers; idempotent graceful close; thread-safe. | [API](client-api.md) §2, [Impl](client-behavior.md) §9–§10 |
+| C15 | Feature gating: never send scalable-topic commands to a broker that does not advertise support. | [Compatibility](compatibility.md) §3 |
+
+## 3. Optional capabilities
+
+Each is independently optional. A client that omits one is conformant; a client that offers one MUST
+satisfy its full specification and gate it on broker support where applicable.
+
+| Capability | Requirement if implemented | Spec |
+|------------|----------------------------|------|
+| **Transactions** | NewTransaction, transactional send/ack, atomic commit/abort; TC discovery (watch where supported, else lookup). | [API](client-api.md) §6, [Wire](wire-protocol.md) §8 |
+| **Namespace subscription** (stream) | Live membership via the namespace watch; cross-topic cumulative ack; establishment-error surfacing. | [API](client-api.md) §4.4, [Wire](wire-protocol.md) §7 |
+| **Entry-bucketing (producer side)** | Single-bucket batching + `entry_hash_min/max` stamping, gated on broker support. A producer that omits it MUST NOT batch on scalable topics. The **consumer side is not optional**: any client offering the Stream consumer MUST implement bucket-shared attach, individual acks, and drain-on-mode-flip ([Key-Shared](key-shared.md) §4, §9). | [Key-Shared](key-shared.md) |
+| **Dead-letter** (queue) | Route messages past the redelivery limit to the dead-letter topic. | [API](client-api.md) §4.2 |
+| **End-to-end encryption** | Producer/consumer encryption per the configured failure action. | [API](client-api.md) §3–§4 |
+| **Chunking / compression / custom batching** | Producer-local policies; no change to delivery semantics. | [API](client-api.md) §3.1 |
+| **Migration bridge** | Accept `persistent://`/short-form names; synthetic-layout + mod-N routing. | [Compatibility](compatibility.md) §1 |
+
+## 4. Non-conformance
+
+A client is **non-conformant** if it: routes keys differently from C2; exposes segments or a partition
+count to the application ([Conventions](conventions.md)); surfaces a not-found as a generic error;
+throws synchronously from an async operation; invalidates a producer/consumer on a transient error; or
+implements an optional capability only partially.
+
+## 5. Test guidance (informative)
+
+Interoperability SHOULD be validated by: routing-vector tests (same key → same segment id across
+clients); a split/merge during active produce/consume (no loss, no reorder, no surfaced error); a
+broker failover (consumer resumes; producer continues); cumulative-ack-across-segments correctness; and
+checkpoint capture/restore round-trips across a layout change.
diff --git a/spec/scalable-topics/conventions.md b/spec/scalable-topics/conventions.md
new file mode 100644
index 000000000000..90f82bf523a3
--- /dev/null
+++ b/spec/scalable-topics/conventions.md
@@ -0,0 +1,80 @@
+# Overview & Conventions
+
+**Status:** Stable
+
+This document defines the scope of the Scalable Topics Client Specification and the conventions every
+other document in the set follows.
+
+## 1. Scope
+
+This specification defines the behavior of a **scalable-topics client SDK**: the producer and consumer
+surface an application uses, the guarantees that surface provides, and the protocol a client speaks to
+an Apache Pulsar broker. It is **language-neutral** — it defines required behavior, not a particular
+language's syntax.
+
+In scope:
+
+- The observable **API contract** (operations, inputs, outputs, ordering and delivery guarantees,
+ errors) — see [Client API](client-api.md).
+- The **implementation requirements** a conformant client MUST meet internally — see
+ [Implementation Requirements](client-behavior.md).
+- The **wire protocol** between client and broker — see [Wire Protocol](wire-protocol.md).
+
+Out of scope: broker-internal behavior (the controller, split/merge execution, storage) except where it
+is observable by a client; and designs whose specification is not yet settled (currently
+geo-replication). A documentable-but-still-changing design MAY appear marked **Experimental** per
+[Stability and Versioning](stability.md). See that document for what is normative vs. out of scope.
+
+## 2. Normative language
+
+The key words **MUST**, **MUST NOT**, **REQUIRED**, **SHALL**, **SHALL NOT**, **SHOULD**,
+**SHOULD NOT**, **RECOMMENDED**, **MAY**, and **OPTIONAL** in this specification are to be interpreted as
+described in [BCP 14](https://www.rfc-editor.org/info/bcp14) ([RFC 2119](https://www.rfc-editor.org/rfc/rfc2119),
+[RFC 8174](https://www.rfc-editor.org/rfc/rfc8174)) when, and only when, they appear in **all
+capitals**.
+
+An implementation is **conformant** if and only if it satisfies every applicable MUST / MUST NOT /
+REQUIRED / SHALL / SHALL NOT requirement. SHOULD-level requirements are strong recommendations whose
+violation requires careful justification; MAY-level items are genuinely optional. The full conformance
+model is in [Conformance](conformance.md).
+
+## 3. Normative vs. informative content
+
+Unless explicitly marked, all content is **normative**. Content marked *“Informative”*, *“Example”*,
+*“Note”*, or *“Rationale”*, and anything in a block quote, is **non-normative** — it aids understanding
+but imposes no requirement. Diagrams are informative.
+
+## 4. Document statuses
+
+Each document and each feature carries a status, defined in [Stability and Versioning](stability.md):
+
+- **Stable** — will not change in backward-incompatible ways without a major specification version bump.
+- **Experimental** — may change in any way; implementations MAY support it but MUST NOT be considered
+ non-conformant for omitting it.
+- **Draft** — not yet ratified; present for review.
+- **Deprecated** — retained for compatibility; SHOULD NOT be used by new code.
+
+## 5. Language-neutral naming and the Java mapping
+
+Operations, types, and configuration are named **language-neutrally** in this specification. Concrete
+SDKs MUST provide these operations but SHOULD name and shape them idiomatically for their language
+(e.g. async style, naming case, error reporting).
+
+The specification uses a small, stable set of operation names — for example *CreateProducer*, *Send*,
+*Subscribe*, *Receive*, *Acknowledge*, *AcknowledgeCumulative*, *Checkpoint*, *NewTransaction*,
+*Commit* — defined where they are introduced. Each document presents normative behavior against these
+names; the **Java V5 reference surface** for every operation and type is collected in
+[Java Reference Mapping](java-mapping.md). Where this specification needs to refer to a concrete symbol
+(a wire command, a metadata field), it uses the on-the-wire / protocol name, which is language-neutral
+by definition.
+
+When a normative requirement is illustrated with a code-like signature, that signature is **illustrative
+pseudocode** unless it appears in [Java Reference Mapping](java-mapping.md) or the
+[Wire Protocol](wire-protocol.md); the prose is the normative form.
+
+## 6. Reference implementation
+
+The Java *V5 client* in `apache/pulsar` is the reference implementation. It is a guide to intended
+behavior, but **this specification is authoritative**: where an implementation (including the reference
+one) diverges from a normative requirement here, that is a defect in the implementation or a defect in
+this specification, to be reconciled — not a silent redefinition of the contract.
diff --git a/spec/scalable-topics/data-model.md b/spec/scalable-topics/data-model.md
new file mode 100644
index 000000000000..6e350017ff36
--- /dev/null
+++ b/spec/scalable-topics/data-model.md
@@ -0,0 +1,107 @@
+# Data Model
+
+**Status:** Stable
+
+This document defines the data objects a scalable-topics client exposes and the identity scheme for
+topics. It is language-neutral; the Java surface is in [Java Reference Mapping](java-mapping.md). Terms
+in *italics* are defined in [Terminology](terminology.md).
+
+## 1. Topic identity
+
+A *scalable topic* is identified by a **canonical name** of the form:
+
+```
+topic://<tenant>/<namespace>/<name>
+```
+
+- A client MUST accept a canonical `topic://` name and use it for every operation on the topic.
+- A client MUST treat the canonical name returned by the broker during topic resolution as
+ authoritative, and SHOULD use it when reporting the topic of a producer, consumer, or message.
+- A scalable topic has **no application-visible partitions**. A client MUST NOT expose a partition count
+ or partition index for a scalable topic.
+
+A client MAY also accept the *migration-bridge* name forms (`persistent://…` and the short form) for
+addressing an existing classic topic as a scalable topic; these are defined in
+[Compatibility](compatibility.md). New applications use `topic://`.
+
+## 2. Key and the hash ring
+
+Each message MAY carry a **key** (a string). The key has two observable effects:
+
+1. It establishes *per-key ordering* (§ Client API): messages with the same key are delivered in
+ publish order to an *ordered consumer*.
+2. It determines the message's *routing* to a *segment* (an internal mechanism defined in
+ [Implementation Requirements](client-behavior.md)).
+
+A message MAY be **keyless**, in which case it carries no ordering guarantee.
+
+The keyspace is the 16-bit **hash ring** `0x0000`–`0xFFFF`. The mapping from key to ring position, and
+from ring position to segment, is specified in [Implementation Requirements](client-behavior.md). The
+hash ring is not otherwise application-visible.
+
+## 3. Message
+
+A **message** received from a scalable topic exposes the following. A client MUST provide read access to
+every field marked REQUIRED and SHOULD provide the others where the underlying transport supplies them.
+
+| Field | Required | Meaning |
+|-------|----------|---------|
+| value | REQUIRED | The deserialized payload, per the *schema*. |
+| raw payload | REQUIRED | The undecoded payload bytes. |
+| message identifier | REQUIRED | The message's *message identifier* (§4). |
+| key | OPTIONAL | The message key, absent if the message is keyless. |
+| properties | REQUIRED | Application-defined string key/value metadata (possibly empty). |
+| publish time | REQUIRED | Broker-assigned publish timestamp. |
+| event time | OPTIONAL | Producer-assigned application timestamp, if set. |
+| sequence identifier | REQUIRED | Producer-assigned sequence number (deduplication). |
+| producer name | OPTIONAL | Name of the publishing producer, if available. |
+| topic | REQUIRED | Canonical topic the message belongs to. |
+| redelivery count | REQUIRED | Number of prior redeliveries (0 on first delivery). |
+| size | REQUIRED | Uncompressed payload size in bytes. |
+| replicated-from | OPTIONAL | Source cluster, if the message arrived via geo-replication. |
+
+A message is **immutable** once delivered.
+
+## 4. Message identifier
+
+A **message identifier** uniquely identifies a message within a topic.
+
+- It MUST be **opaque**: a client MUST NOT expose any internal structure (e.g. ledger, entry, segment,
+ or partition components) through the public identifier type.
+- It MUST be **totally ordered** within a topic: two identifiers from the same topic MUST be comparable,
+ and the order MUST match publish order within a single ordered substream.
+- It MUST be **serializable** to bytes and restorable from those bytes. A serialized identifier produced
+ by one version MUST be restorable by any later version (see [Stability](stability.md)).
+- Two sentinel identifiers MUST be provided: **earliest** (oldest available message) and **latest** (the
+ next message to be published).
+
+A message identifier MUST NOT be used to express a read position spanning multiple segments; a
+*checkpoint* (§5) is used for that.
+
+## 5. Checkpoint
+
+A **checkpoint** is a position value used by the *checkpoint consumer*.
+
+- It MUST be **opaque** and represent a **consistent position across all of a topic's segments** — not a
+ single point in one segment.
+- It MUST be **serializable** to bytes and restorable from those bytes, with the same cross-version
+ forward-compatibility guarantee as a message identifier (see [Stability](stability.md)), because
+ applications persist checkpoints in external state and restore them after upgrades.
+- Two sentinel checkpoints MUST be provided: **earliest** and **latest**.
+- A checkpoint is the **only** position type accepted by the checkpoint consumer; a message identifier
+ MUST NOT be used there.
+
+Timestamp-based positioning is not expressed through the checkpoint type; it is an administrative
+operation (out of scope for this document).
+
+## 6. Schema
+
+Every producer and consumer is created with a **schema** that governs payload serialization and
+deserialization.
+
+- Schema is **topic-level**: a scalable topic has a single schema regardless of how many segments back
+ it. A client MUST present one consistent schema for the topic, not a per-segment schema.
+- A client MUST support the standard Pulsar schema set (bytes, string, the primitive types, key/value,
+ and generic records) as exposed by its schema factory.
+- Schema compatibility MUST be enforced by the broker on connect; a client MUST surface a schema
+ incompatibility as a typed error (see [Error Model](error-model.md)).
diff --git a/spec/scalable-topics/error-model.md b/spec/scalable-topics/error-model.md
new file mode 100644
index 000000000000..f20cf48079db
--- /dev/null
+++ b/spec/scalable-topics/error-model.md
@@ -0,0 +1,83 @@
+# Error Model
+
+**Status:** Stable
+
+This document defines how a scalable-topics client reports failures: the error categories, which are
+retried internally vs. surfaced, and the rules every operation follows. Error *categories* are
+language-neutral; the Java exception types are in [Java Reference Mapping](java-mapping.md).
+
+## 1. General rules
+
+- **Typed errors.** Every fallible operation MUST report failure as a **typed** error category (below),
+ not an opaque/generic failure, so an application can branch on the cause.
+- **Async never throws synchronously.** An operation that returns a future MUST signal all failures by
+ completing the future exceptionally; it MUST NOT throw synchronously from the call that creates the
+ future. (Blocking variants surface the same category as a thrown exception.)
+- **A failed publish/receive does not invalidate the producer/consumer.** Per
+ [Client API](client-api.md) §2, a failed *Send* means that *message* failed; the producer remains
+ valid. The application retries the operation on the same instance.
+- **Internal-only failures are not surfaced.** Transient conditions the client resolves itself
+ (reconnect, layout refresh, segment-gone reroute) MUST NOT reach the application unless retries are
+ exhausted.
+
+## 2. Error categories
+
+| Category | Meaning | Typical origin |
+|----------|---------|----------------|
+| **NotFound** | Topic (or namespace) does not exist and could not be auto-created. | Resolve/subscribe with auto-create disallowed. |
+| **Authentication** | Credentials rejected. | Connect. |
+| **Authorization** | Authenticated but not permitted. | Resolve, produce, subscribe, watch. |
+| **Timeout** | Operation did not complete within its deadline. | Send timeout, lookup/op timeout. |
+| **AlreadyClosed** | The client/producer/consumer is closed. | Any op after close. |
+| **ProducerBusy / Fenced** | Exclusive access mode held by another producer (or this one was fenced). | CreateProducer / send under exclusive. |
+| **ConsumerBusy** | Subscription constraint prevents this consumer attaching. | Subscribe. |
+| **IncompatibleSchema** | Schema not compatible with the topic's. | Producer/consumer connect. |
+| **TopicTerminated** | The target (segment) topic was terminated. | Send to a sealed/migrated segment (internal; see §3). |
+| **Connect / NotConnected** | Transport could not connect or is momentarily disconnected. | Any op (usually internal). |
+| **ProducerQueueFull** | Pending-send queue is full and blocking is disabled. | Send with `blockIfQueueFull = false`. |
+| **MemoryBufferFull** | Client memory limit reached. | Send. |
+| **TransactionConflict** | Transaction operation conflicts/cannot proceed. | Transactional send/ack/commit. |
+| **Crypto** | Encryption/decryption failure (subject to the configured failure action). | Produce/consume on an encrypted topic. |
+| **InvalidConfiguration** | Builder configuration invalid (e.g. topic not set). | Create/subscribe. |
+| **InvalidServiceURL** | Malformed service URL. | Client build. |
+| **Interrupted** | A blocking call was interrupted. | Blocking ops. |
+
+A client MUST surface **NotFound** as a category distinct from generic failure (so "topic absent" is
+distinguishable). For a scalable topic, a not-found result from topic resolution
+([Wire Protocol](wire-protocol.md) §4) MUST map to **NotFound**, not a generic protocol error.
+
+## 3. Retryability
+
+A conformant client MUST classify failures as follows.
+
+**Retried internally (not surfaced unless exhausted):**
+
+- **Connect / NotConnected**, and lookup/assignment/watch connection drops — re-establish with backoff
+ ([Implementation Requirements](client-behavior.md) §8–§9).
+- **TopicTerminated / segment-gone** during producing — drop the per-segment producer, await the new
+ layout, re-route, retry up to a bounded number of attempts
+ ([Implementation Requirements](client-behavior.md) §4). Only exhaustion surfaces (typically as the
+ last underlying error).
+
+**Surfaced to the application (not retried by the client):**
+
+- **NotFound**, **Authentication**, **Authorization**, **IncompatibleSchema**,
+ **InvalidConfiguration**, **InvalidServiceURL** — deterministic; retrying without change cannot help.
+- **ProducerBusy / Fenced**, **ConsumerBusy** — access-mode/subscription constraints.
+- **Timeout**, **ProducerQueueFull**, **MemoryBufferFull**, **TransactionConflict**, **Crypto** —
+ surfaced; the **application** decides whether to retry (the SDK does not retry these automatically).
+
+## 4. Watch establishment errors
+
+Each watch family — DAG watch ([Wire Protocol](wire-protocol.md) §4), namespace watch (§7), TC
+discovery (§8) — signals an establishment failure via `error`/`message` in its first update. A client
+MUST surface that as the failure of the operation that opened the watch (e.g. a namespace watch failing
+with not-found/authorization MUST fail the namespace consumer's *Subscribe*), in the corresponding
+category from §2.
+
+## 5. Negative acknowledgment is not an error
+
+`NegativeAcknowledge` (queue consumer) is a normal control signal requesting redelivery, not an error
+report. A redelivered message carries an incremented redelivery count
+([Data Model](data-model.md) §3); exceeding a dead-letter threshold routes the message to the
+dead-letter topic rather than raising an error.
diff --git a/spec/scalable-topics/java-mapping.md b/spec/scalable-topics/java-mapping.md
new file mode 100644
index 000000000000..3190c8b8e9c8
--- /dev/null
+++ b/spec/scalable-topics/java-mapping.md
@@ -0,0 +1,99 @@
+# Appendix A — Java Reference Mapping
+
+**Status:** Stable · Informative
+
+This appendix maps the language-neutral concepts and operations of this specification to the **Java V5
+client** (`org.apache.pulsar.client.api.v5`), the reference implementation. It is **informative**: the
+normative contract is in the other documents. Other-language SDKs use this only to cross-check intent.
+
+## 1. Entry point and types
+
+| Spec concept | Java type / call |
+|--------------|------------------|
+| Client | `PulsarClient` (`PulsarClient.builder()`) |
+| Producer factory | `PulsarClient.newProducer(Schema<T>)` → `ProducerBuilder<T>` |
+| Stream consumer factory | `PulsarClient.newStreamConsumer(Schema<T>)` → `StreamConsumerBuilder<T>` |
+| Queue consumer factory | `PulsarClient.newQueueConsumer(Schema<T>)` → `QueueConsumerBuilder<T>` |
+| Checkpoint consumer factory | `PulsarClient.newCheckpointConsumer(Schema<T>)` → `CheckpointConsumerBuilder<T>` |
+| Transaction factory | `PulsarClient.newTransaction()` / `newTransactionAsync()` |
+| Message | `Message<T>` |
+| Message identifier | `MessageId` (`toByteArray`/`fromByteArray`, `earliest`/`latest`) |
+| Checkpoint | `Checkpoint` (`toByteArray`/`fromByteArray`, `earliest`/`latest`) |
+| Schema | `org.apache.pulsar.client.api.v5.schema.Schema` |
+| Async views | `async()` → `org.apache.pulsar.client.api.v5.async.Async*` |
+
+## 2. Producer
+
+| Spec | Java |
+|------|------|
+| CreateProducer | `ProducerBuilder.create()` / `createAsync()`; `.topic`, `.producerName`, `.accessMode`, `.sendTimeout`, `.blockIfQueueFull`, `.compressionPolicy`, `.batchingPolicy`, `.chunkingPolicy`, `.encryptionPolicy`, `.initialSequenceId`, `.property/properties` |
+| Send | `producer.newMessage()` → `MessageBuilder<T>`; `.value`, `.key`, `.property/properties`, `.eventTime`, `.sequenceId`, `.deliverAfter/deliverAt`, `.transaction`, `.replicationClusters`, `.send()` → `MessageId` |
+| Last sequence id | `Producer.lastSequenceId()` |
+| Flush / Close | `AsyncProducer.flushAsync()` ; `Producer.close()` |
+| Access modes | `ProducerAccessMode.{SHARED, EXCLUSIVE, EXCLUSIVE_WITH_FENCING, WAIT_FOR_EXCLUSIVE}` |
+
+## 3. Consumers
+
+| Spec | Java |
+|------|------|
+| Stream subscribe | `StreamConsumerBuilder.subscribe()`; `.topic` XOR `.namespace(...)`, `.subscriptionName`, `.subscriptionInitialPosition`, `.acknowledgmentGroupTime`, `.readCompacted`, `.replicateSubscriptionState`, … |
+| Stream receive | `StreamConsumer.receive()` / `receive(Duration)` / `receiveMulti(int, Duration)` |
+| Stream cumulative ack | `StreamConsumer.acknowledgeCumulative(MessageId[, Transaction])` |
+| Queue subscribe | `QueueConsumerBuilder.subscribe()` (topic + subscriptionName) |
+| Queue ack / nack | `QueueConsumer.acknowledge(MessageId[, Transaction])` ; `negativeAcknowledge(MessageId)` |
+| Checkpoint create | `CheckpointConsumerBuilder.create()`; `.topic`, `.startPosition(Checkpoint)`, `.consumerGroup`, … |
+| Checkpoint receive | `CheckpointConsumer.receive()` / `receive(Duration)` / `receiveMulti(int, Duration)` |
+| Capture checkpoint | `CheckpointConsumer.checkpoint()` → `Checkpoint` |
+| Initial position (stream) | `SubscriptionInitialPosition.{Earliest, Latest}` |
+
+> Note: `StreamConsumer` and `CheckpointConsumer` are `Closeable`; `QueueConsumer` is `AutoCloseable`.
+> The grouped/ungrouped checkpoint distinction is `CheckpointConsumerBuilder.consumerGroup(...)` set vs.
+> unset.
+
+## 4. Message
+
+| Spec field | Java `Message<T>` |
+|------------|-------------------|
+| value / raw payload / size | `value()` / `data()` / `size()` |
+| message identifier | `id()` |
+| key | `key()` → `Optional<String>` |
+| properties | `properties()` |
+| publish time / event time | `publishTime()` → `Instant` ; `eventTime()` → `Optional<Instant>` |
+| sequence identifier | `sequenceId()` |
+| producer name / topic | `producerName()` → `Optional<String>` ; `topic()` |
+| redelivery count | `redeliveryCount()` |
+| replicated-from | `replicatedFrom()` → `Optional<String>` |
+
+## 5. Transactions
+
+| Spec | Java `Transaction` |
+|------|--------------------|
+| state | `state()` → `Transaction.State.{OPEN, COMMITTING, ABORTING, COMMITTED, ABORTED, ERROR, TIMED_OUT}` |
+| commit / abort | `commit()` / `abort()` (and `async()`) |
+| transactional publish / ack | `MessageBuilder.transaction(txn)` ; `*Consumer.acknowledge*(id, txn)` |
+
+## 6. Errors
+
+Error categories ([Error Model](error-model.md)) map to `PulsarClientException` (v5) subtypes:
+`NotFoundException`, `AuthenticationException`, `AuthorizationException`, `TimeoutException`,
+`AlreadyClosedException`, `ProducerBusyException`, `ConsumerBusyException`,
+`IncompatibleSchemaException`, `TopicTerminatedException`, `ConnectException`, `NotConnectedException`,
+`ProducerQueueIsFullException`, `MemoryBufferIsFullException`, `TransactionConflictException`,
+`CryptoException`, `InvalidConfigurationException`, `InvalidServiceURLException`.
+
+## 7. Configuration / policies
+
+Spec configuration maps to the `org.apache.pulsar.client.api.v5.config` package:
+`ProducerAccessMode`, `SubscriptionInitialPosition`, `CompressionType`/`CompressionPolicy`,
+`BatchingPolicy`, `ChunkingPolicy`, `ProducerEncryptionPolicy`/`ConsumerEncryptionPolicy`,
+`ProducerCryptoFailureAction`/`ConsumerCryptoFailureAction`, `ConnectionPolicy`, `BackoffPolicy`,
+`TlsPolicy`, `ProxyProtocol`, `TransactionPolicy`, `ProcessingTimeoutPolicy`, `DeadLetterPolicy`,
+`MemorySize`.
+
+## 8. Entry-bucketing
+
+**Entry-bucketing** ([Scalable Key-Shared Consumption](key-shared.md)) has no application-facing Java
+surface — the producer buckets and stamps internally, and the Stream consumer applies bucket-shared
+assignments (Key_Shared attach, individual acks, drain on mode flips) internally. An application only
+observes that creating more Stream consumers than segments increases parallelism. When the Java client
+exposes controls (e.g. per-topic bucket policy), this section will map them.
diff --git a/spec/scalable-topics/key-shared.md b/spec/scalable-topics/key-shared.md
new file mode 100644
index 000000000000..08c8a8214b68
--- /dev/null
+++ b/spec/scalable-topics/key-shared.md
@@ -0,0 +1,181 @@
+# Scalable Key-Shared Consumption (Entry-Bucketing)
+
+**Status:** Stable
+
+> **Normative.** This feature is specified by [PIP-486](../../pip/pip-486.md) and is a **normative**
+> part of this specification — not Experimental. How much of it a client must implement depends on what
+> the client offers (§9): the consumer side is REQUIRED for Stream consumers (the controller may share a
+> segment at any time), while producer-side bucketing is exchangeable for not batching. Where this
+> document and PIP-486 differ, PIP-486 is the source of truth until this document is finalized.
+
+This document overlays the Stable specification with **entry-bucketing**: the mechanism that lets an
+ordered (Stream) subscription scale beyond one consumer per segment with **per-key affinity** (one
+consumer per key), *without* coupling the producer's batching to the consumer's mode.
+
+## 1. Motivation (informative)
+
+The Stable [Stream consumer](client-api.md) scales by segments: each segment has one Exclusive owner, so
+parallelism is capped at the segment count. Classic Pulsar can go finer (`Key_Shared`) but only by
+constraining producer batching, because a batch may mix keys owned by different consumers. Entry-bucketing
+removes that coupling: the producer groups its batches by a small routing tag and stamps it in cleartext
+entry metadata, and the broker routes **whole entries** to a single consumer by that tag — no per-message
+hashing, no decompression, no producer/consumer batching coupling, and no client-side handoff protocol.
+
+## 2. The bucket
+
+A **bucket** is an *intra-segment* routing unit, independent of the segment hash ring.
+
+- A key is hashed **once** with `Murmur3_32` over its UTF-8 bytes; the raw (unmasked) 32-bit hash splits
+ into two independent 16-bit halves: the **high half** routes segments (see
+ [Wire Protocol](wire-protocol.md) §5) and the **low half** is the key's **bucket hash**. Using disjoint
+ bits of one hash is what makes bucket spread independent of the segment's ring-range: the keys a
+ segment receives still cover the full 16-bit bucket-hash space.
+- A segment's buckets are defined by a **boundary list**: ascending, inclusive, contiguous ranges tiling
+ the bucket-hash space `0x0000`–`0xFFFF`. A key belongs to the bucket whose range contains its bucket
+ hash. Ranges need not be equal-width (boundaries MAY be placed to balance buckets by traffic).
+- The boundary list is **chosen per segment** and is **immutable for that segment's life**. Changing it
+ is done by *rebucket rollover* (§5), never by mutating a live segment. It is advertised in the topic
+ layout (`SegmentInfoProto.entry_bucket_splits` — the start hashes of buckets `1..N-1`; an empty list
+ means a single bucket spanning the whole ring).
+- A bucket is owned by **exactly one** consumer at a time within a subscription; this is the source of
+ key affinity. A consumer MAY own several buckets.
+
+The bucket count `N` (the boundary-list size) is decoupled from the topic's segment count: segment
+scaling and bucket scaling are orthogonal axes.
+
+## 3. Producer: entry-bucketing
+
+A producer that implements entry-bucketing MUST:
+
+- Batch such that **every stored entry (batch) belongs to a single bucket** — i.e. group messages by
+ bucket within each destination segment, using the segment's advertised boundary list.
+- Stamp, in the entry's **cleartext outer metadata**, the entry's **effective** bucket-hash range:
+ `MessageMetadata` `entry_hash_min` / `entry_hash_max` — the smallest and largest bucket hash actually
+ present among the entry's messages, both inclusive (16-bit). Because the entry is single-bucket this
+ range necessarily lies within one bucket; it is the *tightest* bound, not the bucket's nominal
+ boundaries. A single-message entry stamps `min = max =` that message's bucket hash.
+- Stamp **always** — including on a single-bucket segment (`N = 1`); the stamp is still the entry's
+ effective min/max, never a synthetic whole-ring range.
+ The stamp records the entry's *effective* hash range at publish time, so it stays meaningful when the
+ entry is later re-examined under a different layout (geo-replication into a cluster with different
+ segment or bucket boundaries, or a rebucketed successor): if the stamped range falls entirely inside
+ one target bucket, the entry still routes whole.
+
+A producer MUST NOT be required to disable or alter batching to enable key-shared consumption; that is
+the entire point. The batching cost of bucketing scales with `1/N`, so a topic with few consumers uses a
+small `N` (good batching) and one feeding many consumers uses a larger `N`.
+
+A producer that does **not** implement bucketing MUST NOT batch on a scalable topic: a multi-key batch
+that straddles buckets is routed whole to one consumer, silently breaking per-key affinity for every
+other consumer of the topic. Publishing each message as its own (possibly unstamped) entry is always
+safe — the broker routes an unstamped entry by the message key's bucket hash.
+
+## 4. Consumer: key-shared dispatch
+
+Entry-bucketing extends the Stream consumer's controller assignment ([Wire Protocol](wire-protocol.md)
+§6.2). Each assigned segment carries a `bucket_ranges` list that selects the **attach mode**:
+
+- **Empty `bucket_ranges`** — the consumer is the segment's **sole owner**: it subscribes **Exclusive**,
+ exactly as in the base protocol. This is the common case and keeps the single-owner fast path (no
+ per-message pending-ack tracking, cumulative acks).
+- **Non-empty `bucket_ranges`** — the segment is **shared by bucket**. The list is the segment's **full
+ boundary list** (§2) — *not* a per-consumer slice — and is **identical for every sharer**. The
+ consumer subscribes `Key_Shared` STICKY with the `KeySharedMeta.entryBucketDispatch` flag, declaring
+ exactly this boundary list in the subscribe's `KeySharedMeta.hashRanges`.
+
+The controller only decides **which consumers share a segment**; the bucket→consumer spread is computed
+**broker-side** from the live membership of the subscription, deterministically (every sharer declares
+the same boundaries, so the broker validates them and rejects a mismatch as a stale-layout error).
+
+**Broker dispatch and handoff (normative for brokers, informative for clients).** The broker routes each
+entry, whole, to the consumer owning its bucket — by the cleartext stamp, with no per-message key hashing
+and no payload decompression. When membership changes (a sharer joins, leaves, or crashes), the broker
+re-spreads buckets and preserves per-key order by **draining**: a moving bucket is withheld from its new
+owner until every message the prior owner has in flight for that bucket is acknowledged, then flows to
+the new owner in order. Consumers keep consuming throughout — they are never rejected, never re-subscribe,
+and implement **no handoff protocol**; a bucket move is observable only as a pause on that bucket.
+
+A client that implements entry-bucketing MUST, in addition:
+
+- **Acknowledge individually on shared segments.** `Key_Shared` forbids cumulative acknowledgment. A
+ client whose consumer API is cumulative (the Stream consumer's position-vector ack,
+ [Implementation Requirements](client-behavior.md) §6) MUST translate it on shared segments by
+ individually acknowledging every delivered-but-unacknowledged message at or before the acked position,
+ per segment. On Exclusive (sole-owner) segments cumulative acknowledgment is used as normal.
+- **Drain before the mode flip.** The only client-visible transition is a segment's `bucket_ranges`
+ flipping between empty and non-empty (sole owner ⇄ shared). Applying it means closing the per-segment
+ consumer and re-subscribing in the other mode; before closing, the client MUST *drain*: stop delivering
+ new messages from that segment and wait until every already-delivered message of the segment is
+ acknowledged. This makes the flip invisible to per-key order and prevents redelivery of prefetched
+ messages. The same drain applies when a segment leaves the assignment entirely. (Draining is bounded by
+ the application acknowledging what it was delivered; a client MUST NOT acknowledge on the application's
+ behalf to force progress.)
+- **Apply assignments idempotently.** Re-receiving an assignment with an unchanged `bucket_ranges` for a
+ segment MUST NOT re-subscribe or otherwise disturb that segment.
+
+The number of consumers that can usefully share a segment is bounded by that segment's bucket count; the
+controller MAY leave surplus consumers idle.
+
+## 5. Changing the boundaries: rebucket rollover
+
+Because a segment's boundary list is immutable, changing it (more buckets, fewer, or re-placed
+boundaries) is performed as a **rebucket rollover**: the controller seals the segment and creates a
+successor with the **same hash range** but a new boundary list. The sealed predecessor drains under its
+old boundaries; the successor takes new writes under the new ones. This reuses the seal → successor →
+producer-redirect flow of an ordinary split (with an unchanged range), so per-key order across the change
+is preserved by the existing mechanism.
+
+A client therefore never observes two different boundary lists for one segment; new boundaries arrive
+only as a new (same-range) successor segment, picked up through the normal layout-change handling.
+
+## 6. The segments-vs-buckets lever (informative)
+
+Total consumer parallelism is `segments × N`, but a lower `N` means fewer, fuller batches.
+Implementations and operators therefore prefer **adding segments** (each at `N = 1`, best batching) and
+raise `N` only when more segments are not the right answer — e.g. at the maximum segment count, for
+low-throughput topics that should not materialize many segments, or to drain a sealed segment's backlog
+across several consumers. The first segment of a new topic MAY default to a small `N > 1` (e.g. 4) to
+allow immediate fan-out before any split.
+
+## 7. Wire additions
+
+Entry-bucketing adds no commands; it adds fields to existing structures (see
+[Wire Protocol](wire-protocol.md)):
+
+- `MessageMetadata.entry_hash_min` / `entry_hash_max` — the producer's bucket stamp (§3). Optional,
+ inert on classic topics, and ignored by brokers without the feature.
+- `SegmentInfoProto.entry_bucket_splits` — the segment's boundary list in the topic layout (§2), so
+ producers can bucket and the broker can validate.
+- `KeySharedMeta.entryBucketDispatch` — the subscribe flag selecting bucket dispatch, with the
+ boundary list carried in the existing `KeySharedMeta.hashRanges` (§4). (These two fields use the
+ proto's camelCase naming, unlike the snake_case fields above.)
+- `ScalableAssignedSegment.bucket_ranges` — the controller's per-segment mode signal in the Stream
+ assignment (§4).
+
+## 8. Encryption
+
+Because routing uses only the cleartext stamp and never reshapes an entry, key-shared consumption works
+for end-to-end-encrypted topics: a single-bucket entry routes to its one owner without the broker
+decrypting or slicing it. (An encrypting producer publishes each message as its own entry — encrypted
+payloads are not batched — but still stamps each entry's bucket range.)
+
+## 9. Conformance
+
+What a client must implement follows from what it offers — the deciding fact is that segment sharing is
+**controller-driven**: the registration carries no capability negotiation, so any Stream consumer may be
+handed a shared segment. (*Informative:* negotiation is deliberately absent — entry-bucketing ships with
+the initial scalable-topics protocol in Pulsar 5.0.0, so no Stream-capable SDK predates it; the consumer
+side is simply part of the Stream consumer contract.)
+
+- **Stream consumer:** a client that offers the Stream consumer MUST implement §4 (mode selection from
+ `bucket_ranges`, individual acks on shared segments, drain before mode flips). The client-side surface
+ is deliberately small — one subscribe flag plus the boundary list, no handoff protocol.
+- **Producer:** a client SHOULD implement §3 (bucketing and stamping). A client that does not MUST NOT
+ batch on scalable topics (§3) — unbatched, unstamped entries remain fully interoperable, at the cost
+ of batching throughput and per-entry key hashing on the broker.
+- **Queue and Checkpoint consumers** are unaffected by entry-bucketing; a client offering only those
+ need implement nothing from this document.
+
+A client MUST gate producer stamping on observed broker support (brokers without the feature ignore the
+fields, so stamping is safe; the gate exists so clients do not bucket-batch — and shrink their batches —
+for brokers that cannot use the stamps).
diff --git a/spec/scalable-topics/stability.md b/spec/scalable-topics/stability.md
new file mode 100644
index 000000000000..0859227eac22
--- /dev/null
+++ b/spec/scalable-topics/stability.md
@@ -0,0 +1,130 @@
+# Stability and Versioning
+
+**Status:** Stable
+
+This document defines how the specification is versioned, how feature maturity is expressed, and what
+compatibility guarantees clients and brokers owe each other.
+
+## 1. Specification versioning
+
+This specification carries a version of the form `MAJOR.MINOR`:
+
+- **MAJOR** increments on a backward-incompatible change to a *Stable* requirement, and is **tied to the
+ Apache Pulsar LTS cadence**: each new MAJOR corresponds to a new Pulsar **LTS** release. Consequently,
+ backward-incompatible changes and removals of Deprecated features (§2.1) land **only at an LTS
+ boundary**, never in an interim release.
+- **MINOR** increments on backward-compatible additions or clarifications, and may occur in any release.
+
+The current version is declared in the [README](README.md). While the specification is at version `0.x`
+it is **Draft**: any part MAY change. Stability guarantees below take full effect from version `1.0`.
+
+A version increment, like any normative change, is enacted through the **change process**: the
+specification is not itself a PIP, and every normative change MUST be made via an accepted PIP whose
+spec edits merge together with it (see [README → Change process](README.md#change-process)).
+
+## 2. Feature stability tiers
+
+Every feature (operation, guarantee, wire command, configuration item) has a stability tier. When a
+document does not state otherwise, the tier is that of its containing document.
+
+- **Stable** — Will not change in a backward-incompatible way without a MAJOR version bump. Conformant
+ clients MUST implement all Stable, non-optional requirements.
+- **Experimental** — May change in any way, including removal, in a MINOR version. Clients MAY implement
+ it; omitting it MUST NOT make a client non-conformant. New, recently-shipped behavior starts here.
+- **Deprecated** — Stable but scheduled for removal. Retained for compatibility; new code SHOULD NOT
+ rely on it. A Deprecated feature MUST NOT be removed before the next MAJOR version.
+
+A feature's tier is independent of its quality; it expresses only the **change contract**.
+
+### 2.1 Deprecation and removal
+
+A feature is moved to **Deprecated** through the [change process](README.md#change-process) (an accepted
+PIP). The deprecating change MUST record:
+
+- the **reason** for deprecation;
+- the **replacement**, or an explicit statement that none exists; and
+- the **earliest specification version** at which removal is permitted.
+
+While a feature is Deprecated:
+
+- the specification MUST mark it **Deprecated** at its point of use, with a pointer to the replacement;
+- a **broker** MUST continue to honor it — the wire forms and behavior of a Deprecated Stable feature
+ remain available until the feature is removed;
+- a **client** SHOULD migrate to the replacement and SHOULD NOT newly adopt the feature;
+- where the deprecation is observable to operators, an implementation SHOULD emit a non-fatal
+ deprecation signal (e.g. a one-time log) rather than failing.
+
+**Removal.**
+
+- A Deprecated **Stable** feature MUST NOT be removed before the next **MAJOR** specification version,
+ and removal MUST go through the change process. The deprecation window therefore spans at least the
+ remainder of the current MAJOR series.
+- Removal MUST also align with the Apache Pulsar **LTS (Long-Term Support)** release cadence: a
+ Deprecated feature MUST remain available at least until the **next LTS release**, so that operators
+ upgrading from one LTS to the next always have a release in which both the deprecated feature and its
+ replacement coexist. A spec MAJOR version that removes Deprecated features is therefore tied to an LTS
+ boundary, not an arbitrary release.
+- An **Experimental** feature has no deprecation protection: it MAY be changed or removed in a MINOR
+ version without a deprecation period (it carries no conformance obligation, §2).
+
+## 3. Scope: committed behavior, plus clearly-marked Experimental designs
+
+This specification describes the **committed behavior** of scalable topics — behavior that is either
+implemented in a released broker, or **ratified via an accepted PIP and committed to ship**. An SDK
+built to it interoperates with current and announced brokers.
+
+A design that is sufficiently specified to document but still **subject to change** MAY be included
+marked **Experimental** (§2), confined to its own clearly-delimited sections. Experimental material
+informs SDK authors of forthcoming behavior but imposes **no conformance obligation** and may change in
+any way, including removal.
+
+On this basis:
+
+- **Scalable key-shared consumption (entry-bucketing, [PIP-486](../../pip/pip-486.md))** is **normative**
+ (ratified via PIP-486). Its consumer side is REQUIRED for clients offering the Stream consumer;
+ producer-side bucketing is exchangeable for unbatched publishing — see
+ [its conformance section](key-shared.md). It is not Experimental.
+- **Geo-replication of scalable topics** is **out of scope** — its design is not yet settled (not even
+ Experimental).
+- No feature is currently in the Experimental tier; the tier remains defined for future use.
+
+## 4. Compatibility guarantees
+
+### 4.1 Wire protocol
+
+- The protocol is **extensible by addition**: new optional fields and new commands MAY be added in a
+ MINOR version. Clients and brokers MUST ignore unknown optional fields.
+- A client MUST NOT assume the presence of an Experimental command or field unless it has negotiated or
+ observed broker support for it.
+- A broker MUST continue to accept the wire forms of every Stable command across MINOR versions.
+
+### 4.2 Client API contract
+
+- The observable guarantees in [Client API](client-api.md) marked Stable MUST hold across MINOR
+ versions. New operations MAY be added; existing guarantees MUST NOT be weakened without a MAJOR bump.
+
+### 4.3 Serialized formats
+
+- A *message identifier* and a *checkpoint* produced by one version MUST be deserializable by any later
+ version (forward-compatible restore). This is REQUIRED because applications persist these values
+ externally and restore them after upgrades.
+
+### 4.4 Broker / client version skew
+
+- A scalable topic (`topic://` / `segment://`) is served only by brokers that implement scalable topics;
+ the new protocol commands therefore never reach a broker that does not understand them. Optional
+ message-metadata additions for scalable topics are inert on classic topics and on brokers without the
+ feature.
+
+## 5. Current feature status
+
+| Area | Tier | Notes |
+|------|------|-------|
+| Topic identity & migration bridge | Stable | |
+| Producer & routing | Stable | |
+| Stream / Queue / Checkpoint consumers | Stable | |
+| Transactions | Stable | Coordinator discovery via assignment watch. |
+| Namespace subscription | Stable | |
+| Wire protocol (`CommandScalableTopic*`, watches) | Stable | |
+| Scalable key-shared consumption (entry-bucketing) | Normative | Ratified via [PIP-486](../../pip/pip-486.md). Consumer side required for Stream consumers; producer side exchangeable for unbatched publishing. |
+| Geo-replication of scalable topics | Out of scope | Design not yet settled. |
diff --git a/spec/scalable-topics/terminology.md b/spec/scalable-topics/terminology.md
new file mode 100644
index 000000000000..9d7764e4f090
--- /dev/null
+++ b/spec/scalable-topics/terminology.md
@@ -0,0 +1,167 @@
+# Terminology
+
+**Status:** Stable
+
+This document gives the normative definitions of terms used throughout the specification. A term in
+*italics* elsewhere in the spec refers to a definition here. Definitions are normative; examples are
+informative.
+
+### Scalable topic
+
+A durable, ordered-by-key message log identified by a *topic identity* (`topic://…`). It is the unit an
+application produces to and consumes from. A scalable topic has **no application-visible partitioning**;
+its only structural concept exposed to applications is the *key*.
+
+### Topic identity
+
+The canonical name of a scalable topic, of the form `topic://<tenant>/<namespace>/<name>`. The broker
+resolves alternate input forms (see *migration bridge*) to this canonical identity.
+
+### Key
+
+An application-supplied string attached to a message that determines the message's *routing* and its
+*per-key ordering*. A message MAY be keyless.
+
+### Hash ring
+
+The 16-bit key-hash space `0x0000`–`0xFFFF`. A *key* is mapped onto the ring by a defined hash function
+(see [Implementation Requirements](client-behavior.md)). *Segments* own contiguous sub-ranges of the
+ring.
+
+### Segment
+
+A broker-internal unit that owns a contiguous sub-range of the *hash ring* and stores the messages whose
+key-hash falls in that range. Segments are an implementation and scaling detail; they are **not** part
+of the application-visible model and MUST NOT be surfaced by the API as an application concept. They are
+visible only at the *wire protocol* level and within the *checkpoint* model.
+
+### Entry-bucket
+
+An intra-*segment* routing unit ([Key-Shared](key-shared.md)): a sub-range of the 16-bit entry-bucket
+hash space, defined by a per-segment boundary list that is immutable for the segment's life. A key maps
+to a bucket by the **low** 16 bits of the same raw Murmur3 hash whose **high** 16 bits place it on the
+*hash ring*. Each bucket is owned by exactly one consumer at a time within a subscription, which is what
+lets several ordered consumers share one segment with per-key affinity.
+
+### Segment state
+
+A segment is **active** (currently accepting writes for its range) or **sealed** (closed by a split or
+merge; still readable until fully *drained*). New writes for a range always target the single active
+segment covering it.
+
+### DAG (segment DAG / layout)
+
+The directed acyclic graph of a scalable topic's segments at a point in time, including each segment's
+range, state, and split/merge lineage. The DAG is **per-cluster and dynamic**: it changes as segments
+split and merge. It carries a monotonically increasing *epoch*.
+
+### Epoch
+
+A monotonically increasing version number for a topic's *DAG*. Used to order and discard stale layout
+updates.
+
+### Split / merge
+
+The operations by which the broker changes the *DAG*: a **split** divides one segment's range across two
+new active successor segments and seals the original; a **merge** combines two adjacent segments. These
+preserve *per-key ordering* across the boundary.
+
+### Per-key ordering
+
+The guarantee that all messages published with the same *key* are delivered to an *ordered consumer* in
+publish order. Because a key maps to one segment lineage, the order is well-defined. Keyless messages
+carry no ordering guarantee.
+
+### Routing
+
+The client-side selection, per message, of the destination *segment*, derived from the message's *key*
+(or round-robin when keyless). Defined in [Implementation Requirements](client-behavior.md).
+
+### Producer
+
+A client object that publishes messages to one scalable topic.
+
+### Consumer
+
+A client object that reads messages from one scalable topic (or, for the *namespace subscription*, a set
+of topics). Three modes exist: *stream consumer*, *queue consumer*, *checkpoint consumer*.
+
+### Stream consumer
+
+An ordered, broker-managed consumer: messages are delivered in *per-key* order with **cumulative**
+acknowledgment, and load is shared across the consumers of a subscription (Failover-like behavior).
+
+### Queue consumer
+
+A parallel, broker-managed consumer: messages are distributed across the consumers of a subscription for
+parallel processing with **no ordering and no key affinity**, using **individual** acknowledgment and
+negative acknowledgment.
+
+### Checkpoint consumer
+
+An unmanaged consumer with no broker-managed cursor: read position is held entirely by the application
+as a *checkpoint*. Intended for connector frameworks. MAY be ungrouped (reads all of the topic) or
+grouped (shares the topic with other members of a *consumer group*).
+
+### Subscription
+
+A named, broker-managed read position (cursor) shared by the consumers reading a scalable topic under
+that name. Used by *stream* and *queue* consumers; not used by *checkpoint* consumers.
+
+### Consumer group (checkpoint)
+
+A named set of *checkpoint consumers* that share a topic's segments via the broker's coordinator without
+a persisted cursor; segments rebalance as members join or leave.
+
+### Cumulative acknowledgment
+
+An acknowledgment that marks every message up to and including a given one as processed. Used by
+*stream consumers*.
+
+### Individual acknowledgment
+
+An acknowledgment that marks exactly one message as processed. Used by *queue consumers*; paired with
+**negative acknowledgment**, which requests redelivery of one message.
+
+### Message identifier
+
+An opaque, serializable, totally-ordered identifier for a message within a topic. It exposes no internal
+structure. It is not used to express a position across multiple segments — see *checkpoint*.
+
+### Checkpoint
+
+An opaque, serializable value representing a consistent read position across **all** of a topic's
+segments. Created by a *checkpoint consumer*, stored by the application, and used to resume.
+
+### Affinity (key affinity)
+
+The property that, within one subscription, at most one consumer handles a given *key* at a time.
+Provided by *ordered* consumption modes; **not** provided by the *queue consumer*.
+
+### Cursor
+
+A broker-managed, persisted read position for a *subscription*.
+
+### Migration bridge
+
+The mechanism by which an existing classic (`persistent://`) topic is addressed as a scalable topic: the
+broker wraps it in a synthetic single-segment *DAG*. A *legacy segment* names the wrapped classic topic
+rather than a `segment://` topic. Provided only to ease migration.
+
+### Legacy segment
+
+A segment in a synthetic layout that wraps an externally-managed `persistent://` topic instead of a
+broker-managed `segment://` topic. See *migration bridge*.
+
+### Controller (leader)
+
+The broker-internal, per-topic authority that manages the *DAG* and assigns segments to *ordered
+consumers*. A client discovers and connects to it as part of the consumer-assignment protocol; it is not
+otherwise application-visible.
+
+### Replicator / replicated message
+
+In geo-replication, the broker component that re-publishes a topic's messages to other clusters. A
+**replicated message** is one received via replication; it is marked with its source cluster.
+(Geo-replication of scalable topics is out of scope for this specification version — see
+[Stability and Versioning](stability.md).)
diff --git a/spec/scalable-topics/wire-protocol.md b/spec/scalable-topics/wire-protocol.md
new file mode 100644
index 000000000000..0d7edcd8f652
--- /dev/null
+++ b/spec/scalable-topics/wire-protocol.md
@@ -0,0 +1,397 @@
+# Wire Protocol
+
+**Status:** Stable
+
+This document specifies the **complete** client↔broker protocol for scalable topics: every command, its
+fields, and every interaction (with sequence diagrams). It binds the transport-agnostic
+[Client API](client-api.md) to the Apache Pulsar binary protocol.
+
+All commands are defined in `pulsar-common/src/main/proto/PulsarApi.proto` and are carried inside
+`BaseCommand`. Standard Pulsar framing, `CommandConnect`/auth, lookup, and the
+producer/consumer/`Send`/`Ack` commands are **reused unchanged** except where noted; this document
+specifies only the scalable-topic-specific exchanges and how the reused ones are applied at the
+**segment** level.
+
+Conventions: `C→B` = client to broker, `B→C` = broker to client. Field names match the proto. A field
+shown `field?` is optional.
+
+## 1. Command catalog
+
+The scalable-topic command family occupies `BaseCommand` field numbers 70–82:
+
+| Command | # | Direction | Purpose |
+|---------|---|-----------|---------|
+| `CommandScalableTopicLookup` | 70 | C→B | Open a DAG-watch session; resolve a topic. |
+| `CommandScalableTopicUpdate` | 71 | B→C | Initial layout **and** every subsequent pushed layout change. |
+| `CommandScalableTopicClose` | 72 | C→B | Close a DAG-watch session. |
+| `CommandScalableTopicSubscribe` | 73 | C→B | Register an ordered (Stream) / external (Checkpoint) consumer; request assignment. |
+| `CommandScalableTopicSubscribeResponse` | 74 | B→C | Initial segment assignment, or error. |
+| `CommandScalableTopicAssignmentUpdate` | 75 | B→C | Push a new assignment after a rebalance. |
+| `CommandWatchScalableTopics` | 76 | C→B | Open a namespace membership watch. |
+| `CommandWatchScalableTopicsUpdate` | 77 | B→C | Snapshot or diff of the matching topic set, or error. |
+| `CommandWatchScalableTopicsClose` | 78 | C→B | Close a namespace watch. |
+| `CommandWatchTcAssignments` | 79 | C→B | Open the transaction-coordinator discovery watch. |
+| `CommandWatchTcAssignmentsUpdate` | 80 | B→C | Full TC `partition → leader` snapshot, or error. |
+| `CommandWatchTcAssignmentsClose` | 81 | C→B | Close the TC discovery watch. |
+| `CommandScalableTopicUnsubscribe` | 82 | C→B | Cleanly leave a subscription; the controller unregisters and rebalances immediately. Acked with `CommandSuccess`. |
+
+Reused (standard) commands applied at the segment level: `CommandProducer`/`CommandSend`,
+`CommandSubscribe`/`CommandFlow`/`CommandMessage`/`CommandAck`, the reader commands, the transaction
+commands (`CommandNewTxn`, `CommandAddPartitionToTxn`, `CommandAddSubscriptionToTxn`, `CommandEndTxn`
+and the `…OnPartition`/`…OnSubscription` variants), and schema commands.
+
+## 2. Shared structures
+
+```
+ScalableTopicDAG {
+ uint64 epoch // monotonic layout version
+ repeated SegmentInfoProto segments
+ repeated SegmentBrokerAddress segment_brokers // segment_id -> broker_url[/_tls]
+ string controller_broker_url[/_tls] // controller leader for this topic
+}
+SegmentInfoProto {
+ uint64 segment_id
+ uint32 hash_start, hash_end // inclusive 16-bit range owned by the segment
+ SegmentState state // ACTIVE | SEALED
+ repeated uint64 parent_ids, child_ids // split/merge lineage (DAG edges)
+ uint64 created_at_epoch [, sealed_at_epoch] // DAG generation numbers
+ uint64 created_at_ms [, sealed_at_ms] // wall clock (retention GC, timestamp seek)
+ string legacy_topic_name? // set => wraps an external persistent:// topic (migration)
+ repeated uint32 entry_bucket_splits // PIP-486 (field 12): ascending inclusive start hashes of
+ // buckets 1..N-1; empty => one bucket spanning the ring
+}
+```
+
+A client MUST treat `epoch` as the monotonic version of a layout and ignore any update whose epoch is
+not greater than the last applied (guards against reordered/stale pushes). The set of `ACTIVE` segments
+whose `[hash_start, hash_end]` ranges tile `0x0000`–`0xFFFF` is the current write surface; `SEALED`
+segments remain readable until drained.
+
+## 3. Segment naming
+
+A managed segment's `segment://` topic name is computed from the parent topic and a descriptor:
+
+```
+segment://<tenant>/<namespace>/<parent-local-name>/<hexStart>-<hexEnd>-<segmentId>
+descriptor = format("%04x-%04x-%d", hash_start, hash_end, segment_id)
+e.g. segment://tenant/ns/my-topic/0000-ffff-0 (full range, single segment)
+ segment://tenant/ns/my-topic/0000-7fff-1 (lower half)
+```
+
+For a **legacy** segment (`legacy_topic_name` set — the synthetic layout wrapping a `persistent://`
+topic during migration) there is no `segment://` name; the client targets the wrapped `persistent://`
+name directly.
+
+## 4. Topic resolution and the DAG-watch session
+
+A producer or queue consumer opens a **DAG-watch session** to resolve the topic and receive its layout
+plus live updates. The session is identified by a client-assigned `session_id`.
+
+```
+CommandScalableTopicLookup { session_id, topic }
+CommandScalableTopicUpdate { session_id, dag?: ScalableTopicDAG, error?, message?, resolved_topic_name? }
+CommandScalableTopicClose { session_id }
+```
+
+- `topic` accepts `topic://`, `persistent://`, or short form; the broker normalizes it and returns the
+ canonical identity in `resolved_topic_name` (set on every success update). The client MUST use that
+ downstream.
+- The **same** `CommandScalableTopicUpdate` is used for the initial response and for every pushed change.
+ The broker streams a new one on every layout change (split/merge, broker reassignment).
+- **Establishment can fail**: if resolution fails (topic not found and auto-create disallowed,
+ authorization denied), the first update sets `error`/`message` with no `dag`; the client MUST surface
+ this as a create/subscribe failure.
+- On connection loss the client re-sends `CommandScalableTopicLookup` (with backoff) to re-establish the
+ watch and refresh the layout before resuming.
+
+```mermaid
+sequenceDiagram
+ participant C as Client
+ participant B as Broker (lookup/owner)
+ C->>B: CommandScalableTopicLookup{session_id, topic}
+ alt resolved
+ B-->>C: CommandScalableTopicUpdate{session_id, dag(epoch=e0), resolved_topic_name}
+ Note over C: cache layout, epoch=e0
+ B-->>C: CommandScalableTopicUpdate{session_id, dag(epoch=e1)} (split/merge)
+ Note over C: apply iff e1 > e0
+ else not found / unauthorized
+ B-->>C: CommandScalableTopicUpdate{session_id, error, message}
+ Note over C: fail create()/subscribe()
+ end
+ C->>B: CommandScalableTopicClose{session_id}
+```
+
+## 5. Producing
+
+The producer resolves the layout (§4), then for each message selects the target **segment** with the
+client-side router and produces to it with the standard `CommandProducer`/`CommandSend` flow.
+
+**Routing (client-side — MUST match exactly so all clients route a key identically):**
+
+1. **Keyed:** `hash = raw_Murmur3_32(key_utf8) >>> 16` — the **high** 16 bits of the raw (unmasked)
+ 32-bit Murmur3 hash; select the single `ACTIVE` segment whose `[hash_start, hash_end]` contains
+ `hash`. (The raw hash is used so the high half is full-range; the **low** 16 bits are reserved as
+ the key's *entry-bucket* hash — [Key-Shared](key-shared.md) §2 — so the two routings are
+ independent.)
+2. **Keyless:** round-robin across the active segments.
+3. **All-legacy (synthetic) layout:** `signSafeMod(Murmur3_32(key_utf8), N)` over the `N` legacy
+ segments (identical to v4 partitioned-topic routing, preserving key→partition while migrating).
+
+The per-segment `CommandProducer` is opened lazily on first send to a segment (named
+`<producerName>-seg-<segment_id>`), against the owning broker from `segment_brokers`. When a pushed
+update **seals** the target segment, the client MUST drop that per-segment producer, re-route to the
+active successor, and continue — preserving per-key order (see the transparent retry in
+[Implementation Requirements](client-behavior.md)).
+
+```mermaid
+sequenceDiagram
+ participant P as Producer (client)
+ participant S0 as Broker for seg-0
+ participant S1 as Broker for seg-1 (successor)
+ Note over P: layout known (§4) — seg-0 ACTIVE covers key K
+ P->>S0: CommandProducer{ segment://…/0000-ffff-0 }
+ P->>S0: CommandSend{ msg(key=K) }
+ S0-->>P: SendReceipt(msgId) — MessageId encodes segment-0
+ Note over P: pushed CommandScalableTopicUpdate: seg-0 SEALED, seg-1 ACTIVE
+ P->>P: close seg-0 producer, re-route K → seg-1
+ P->>S1: CommandProducer{ segment://…/0000-ffff-1 }
+ P->>S1: CommandSend{ msg(key=K) }
+ S1-->>P: SendReceipt(msgId)
+```
+
+## 6. Consuming
+
+### 6.1 Queue consumer — attach to all segments (no assignment protocol)
+
+A queue consumer is driven by the DAG watch (§4). It opens a standard **`CommandSubscribe` (Shared)**
+against **every** active *and* sealed segment topic, multiplexes deliveries, and individually acks
+(routing each ack to the per-segment consumer via the segment id encoded in the `MessageId`). As the
+layout changes it adds/removes per-segment subscriptions.
+
+```mermaid
+sequenceDiagram
+ participant Q as QueueConsumer (client)
+ participant B as Brokers (per segment)
+ Q->>B: CommandScalableTopicLookup{session_id, topic}
+ B-->>Q: CommandScalableTopicUpdate{ dag: seg-0(ACTIVE), seg-1(SEALED) }
+ par attach to every segment
+ Q->>B: CommandSubscribe{ segment://…/0, Shared, sub }
+ Q->>B: CommandSubscribe{ segment://…/1, Shared, sub }
+ end
+ B-->>Q: CommandMessage (multiplexed)
+ Q->>B: CommandAck{ msgId } (routed to owning segment)
+ Note over Q: on layout change, add/remove per-segment subscriptions
+```
+
+### 6.2 Stream / grouped-checkpoint consumer — controller assignment
+
+Ordered (Stream) and grouped-Checkpoint consumers register with the **controller leader** and are
+handed an explicit segment assignment. The controller leader's URL comes from the DAG layout
+(`controller_broker_url`); the client connects to it **directly** (scalable-topic URIs are not
+resolvable through the standard v4 lookup service).
+
+```
+CommandScalableTopicSubscribe { request_id, topic, subscription, consumer_name, consumer_id,
+ consumer_type: STREAM | CHECKPOINT }
+CommandScalableTopicSubscribeResponse { request_id, assignment?: ScalableConsumerAssignment, error?, message? }
+CommandScalableTopicAssignmentUpdate { consumer_id, assignment: ScalableConsumerAssignment }
+
+ScalableConsumerAssignment { uint64 layout_epoch; repeated ScalableAssignedSegment segments }
+ScalableAssignedSegment { uint64 segment_id; uint32 hash_start, hash_end; string segment_topic;
+ repeated IntRange bucket_ranges }
+```
+
+- `consumer_id` is client-chosen and tags every subsequent pushed `CommandScalableTopicAssignmentUpdate`;
+ `request_id` matches the subscribe response.
+- The client attaches to exactly the `segment_topic`s in its current assignment — **Stream** via a
+ standard `CommandSubscribe`, **grouped Checkpoint** via a Reader — and detaches from segments
+ removed by a later assignment.
+- `bucket_ranges` selects the Stream consumer's attach mode per segment (entry-bucketing,
+ [Key-Shared](key-shared.md) §4): **empty** ⇒ sole owner, subscribe **Exclusive**; **non-empty** ⇒ the
+ segment is shared by entry-bucket, and the list is the segment's full bucket-boundary list — the
+ consumer subscribes `Key_Shared` STICKY with `KeySharedMeta.entryBucketDispatch = true` and exactly
+ this list in `KeySharedMeta.hashRanges`. The controller may share a segment whenever consumers
+ outnumber segments, so a Stream client MUST implement this handling. Before applying a mode flip (or
+ dropping a segment) the client MUST drain the segment — see [Key-Shared](key-shared.md) §4.
+- `layout_epoch` lets the client **reject stale** assignments (apply only if epoch ≥ last applied).
+- Rebalances are pushed when a peer joins/leaves the subscription or on a segment split/merge.
+
+```mermaid
+sequenceDiagram
+ participant S as Stream/CkptConsumer (client)
+ participant L as Controller leader
+ participant Seg as Segment brokers
+ S->>L: CommandScalableTopicLookup → controller_broker_url
+ S->>L: CommandScalableTopicSubscribe{request_id, topic, sub, consumer_id, STREAM}
+ alt success
+ L-->>S: CommandScalableTopicSubscribeResponse{request_id, assignment(epoch=e0)=[seg-0]}
+ S->>Seg: CommandSubscribe{ segment://…/0, Exclusive, sub }
+ L-->>S: CommandScalableTopicAssignmentUpdate{consumer_id, assignment(epoch=e1)=[seg-0, seg-1]}
+ Note over S: e1>e0 → attach seg-1 (detach any removed)
+ S->>Seg: CommandSubscribe{ segment://…/1, Exclusive, sub }
+ else error
+ L-->>S: CommandScalableTopicSubscribeResponse{request_id, error, message}
+ end
+```
+
+**Reconnect / grace.** On connection loss after the initial assignment, the client re-runs
+lookup → connect-to-controller → register → subscribe with backoff. Within the controller's grace
+period the broker re-attaches the existing registration and returns the **same** assignment (no
+listener-visible change); past grace, the controller rebalances and the client applies the diff (detach
+removed, attach added).
+
+**Clean leave.** Closing a consumer MUST send `CommandScalableTopicUnsubscribe { request_id,
+consumer_id }` on the registration's connection: the controller deletes the registration and
+rebalances the remaining consumers immediately, instead of holding the session for the grace period.
+This is REQUIRED, not an optimization — the controller connection is pooled, so a consumer's close is
+otherwise invisible while its client process lives (no `channelInactive`), and the registration would
+linger indefinitely. The command is acknowledged with `CommandSuccess` and is idempotent (an unknown
+`consumer_id` — e.g. already swept by a disconnect — still succeeds); the client treats it as
+best-effort and MUST NOT fail the close if it errors, since the grace period remains the fallback for
+unclean departures.
+
+### 6.3 Ungrouped checkpoint consumer
+
+No registration: the client reads **all** segments from its start position directly with per-segment
+Readers (driven by the DAG watch), tracking positions locally; a *checkpoint* serializes the per-segment
+read positions of its segment set.
+
+## 7. Namespace / multi-topic watch
+
+A namespace-subscribing Stream consumer discovers its topic set via a membership watch.
+
+```
+CommandWatchScalableTopics { watch_id, namespace, property_filters[], current_hash? }
+CommandWatchScalableTopicsUpdate { watch_id, oneof{ snapshot: ScalableTopicsSnapshot | diff: ScalableTopicsDiff }, error?, message? }
+CommandWatchScalableTopicsClose { watch_id }
+```
+
+- `property_filters` are AND filters over topic properties; empty ⇒ all topics in the namespace.
+- The subscribe opens a **server-side stream of updates** (not request/response). **Establishment can
+ fail** — namespace not found, authorization denied — in which case the first update sets
+ `error`/`message` and the client MUST fail `subscribe()`.
+- On success the first update carries a `ScalableTopicsSnapshot` (full set; the client replaces its
+ local set); subsequent changes are a `ScalableTopicsDiff` (apply `removed` before `added`).
+- `current_hash` (CRC32C over sorted topic names, same function as `CommandGetTopicsOfNamespace`) is
+ sent on reconnect; if it matches the broker's freshly computed hash, the broker skips re-sending the
+ snapshot.
+
+For **each** topic in the set the client runs an independent per-topic DAG watch (§4) and assignment
+(§6.2), multiplexing into one receive queue; cumulative ack uses a position vector spanning
+`{topic → {segmentId → messageId}}`. On a diff: subscribe **added** topics; for **removed**, detach the
+per-topic consumer without acknowledging — acks are always explicit ([Client Behavior](client-behavior.md) §6).
+
+```mermaid
+sequenceDiagram
+ participant N as NamespaceConsumer (client)
+ participant B as Broker
+ participant T as Per-topic consumers
+ N->>B: CommandWatchScalableTopics{watch_id, namespace, filters, current_hash?}
+ alt established
+ B-->>N: CommandWatchScalableTopicsUpdate{ snapshot=[t1, t2] }
+ N->>T: subscribe t1, t2 (each: §4 DAG watch + §6.2 assignment)
+ B-->>N: CommandWatchScalableTopicsUpdate{ diff: added=[t3], removed=[t1] }
+ N->>T: subscribe t3, detach t1 (no ack)
+ else error
+ B-->>N: CommandWatchScalableTopicsUpdate{ error, message }
+ Note over N: fail subscribe()
+ end
+ N->>B: CommandWatchScalableTopicsClose{watch_id}
+```
+
+## 8. Transactions
+
+Transaction **lifecycle** uses the standard transaction commands; only **coordinator discovery** is
+scalable-topic-specific.
+
+### 8.1 Coordinator discovery — watch, not lookup
+
+Against a broker advertising `supports_tc_metadata_discovery`, the client opens **one** assignment watch
+instead of per-coordinator lookups.
+
+```
+CommandWatchTcAssignments { watch_id }
+CommandWatchTcAssignmentsUpdate { watch_id, snapshot?: TcAssignmentsSnapshot, error?, message? }
+CommandWatchTcAssignmentsClose { watch_id }
+
+TcAssignmentsSnapshot { uint32 parallelism; repeated TcAssignment assignments }
+TcAssignment { uint64 tc_id; string broker_service_url[/_tls] }
+```
+
+- The broker replies with the **full** `partition → leader` map and re-pushes the **whole** snapshot on
+ every leadership change (no diffs, no point lookup). The client replaces its map wholesale.
+- `tc_id` is the coordinator partition (`TxnID.mostSigBits`); the client keeps one handler per partition
+ pointed at its current leader, retargeted on each snapshot.
+- A partition **mid-election** is absent from the snapshot; a transaction routed to it is **parked**
+ until a later snapshot fills it in.
+
+```mermaid
+sequenceDiagram
+ participant C as Client (txn)
+ participant B as Any broker (service URL)
+ participant TC as Coordinator leaders
+ C->>B: CommandWatchTcAssignments{watch_id}
+ B-->>C: CommandWatchTcAssignmentsUpdate{ snapshot{parallelism, [tc0→brokerA, tc1→brokerB]} }
+ Note over C: one handler per tc_id, pointed at its leader
+ B-->>C: CommandWatchTcAssignmentsUpdate{ snapshot{ tc1→brokerC } } (leader moved)
+ Note over C: retarget tc1 handler to brokerC
+```
+
+### 8.2 Lifecycle (standard commands, at segment level)
+
+```mermaid
+sequenceDiagram
+ participant C as Client
+ participant TC as Coordinator (for partition)
+ participant Seg as Segment brokers / subscriptions
+ C->>TC: CommandNewTxn → TxnID
+ Note over C: as the txn touches topics:
+ C->>TC: CommandAddPartitionToTxn{ segment topic produced to }
+ C->>TC: CommandAddSubscriptionToTxn{ subscription acked on }
+ C->>Seg: CommandSend(txn) / CommandAck(txn) (buffered)
+ C->>TC: CommandEndTxn{ COMMIT | ABORT }
+ TC->>Seg: CommandEndTxnOnPartition / CommandEndTxnOnSubscription
+ TC-->>C: EndTxnResponse
+```
+
+A transaction's *partition* set is the set of **segment topics** it actually wrote to; a split/merge
+during an open transaction simply means later writes add the successor segment as another partition.
+
+## 9. Auto-creation
+
+Auto-creation is not a separate command — it happens inside topic resolution (§4) / subscribe (§6).
+When the topic does not exist, the broker creates it (single initial segment) **iff** policy allows and
+then returns the layout/assignment; otherwise it returns `error = not-found`. The client surfaces the
+outcome as the result of `create()`/`subscribe()`.
+
+```mermaid
+sequenceDiagram
+ participant C as Client
+ participant B as Broker
+ C->>B: CommandScalableTopicLookup{session_id, topic (absent)}
+ alt auto-create allowed
+ Note over B: create topic + 1 initial segment
+ B-->>C: CommandScalableTopicUpdate{ dag(1 segment), resolved_topic_name }
+ else disallowed
+ B-->>C: CommandScalableTopicUpdate{ error=not-found, message }
+ end
+```
+
+## 10. Reused subsystems
+
+`CommandConnect`/auth, TLS, schema negotiation (`CommandGetSchema`/`CommandGetOrCreateSchema`), broker
+lookup, and `CommandSend`/`CommandAck` framing are the standard Pulsar protocol, applied at the
+**segment** topic level (`segment://`, or the wrapped `persistent://` for legacy segments). No
+scalable-topic-specific changes are required to them.
+
+## 11. Cross-cutting requirements
+
+- **Epoch monotonicity.** For every layout (`ScalableTopicDAG.epoch`) and assignment
+ (`ScalableConsumerAssignment.layout_epoch`), a client MUST apply an update only if its epoch is ≥ the
+ last applied, discarding stale/reordered pushes.
+- **Watch establishment errors.** Each of the three watch families (DAG §4, namespace §7, TC §8) reports
+ an establishment failure via `error`/`message` in its first update; a client MUST surface it as the
+ failure of the operation that opened the watch.
+- **Reconnect.** Each session (DAG watch, assignment, namespace watch, TC watch) re-establishes with
+ backoff on connection loss and re-syncs via a fresh snapshot/layout; in-progress producers/consumers
+ MUST NOT be invalidated (see [Client API](client-api.md) §2).