From 386845ecd0e75c11b7cd9627bcc91eed5ce9730a Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 5 May 2026 17:24:36 -0700 Subject: [PATCH 1/3] [PIP] PIP-471: Metadata-Driven Transactions for Scalable Topics Sub-PIP of PIP-460 (Scalable Topics). Defines transactional support for scalable topics by replacing the in-stream commit/abort marker model with metadata-store-backed state. Adds parallel implementations of TransactionBuffer, PendingAckStore, and Transaction Coordinator that write nothing to any data stream, so sealed segments (split/merge) no longer strand in-flight transactions. Reuses dispatcher, client API, and TC wire commands; legacy in-stream-marker components remain unchanged for persistent:// topics. --- pip/pip-471.md | 401 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 401 insertions(+) create mode 100644 pip/pip-471.md diff --git a/pip/pip-471.md b/pip/pip-471.md new file mode 100644 index 0000000000000..2ef59c0c27a15 --- /dev/null +++ b/pip/pip-471.md @@ -0,0 +1,401 @@ +# PIP-471: Metadata-Driven Transactions for Scalable Topics + +*Sub-PIP of [PIP-460: Scalable Topics](pip-460.md)* + +## Background + +### Pulsar's existing transaction model + +Pulsar transactions today are realized through three components: + +- **Transaction Coordinator (TC)** — a per-broker service backed by a system topic (`__transaction_log_*` in `pulsar/system`) that tracks the lifecycle of every transaction (`OPEN`, `COMMITTING`, `COMMITTED`, `ABORTING`, `ABORTED`, `TIME_OUT`) and orchestrates two-phase commit across the topics that participate in each transaction. +- **TransactionBuffer (TB)** — a per-`PersistentTopic` component that buffers transactional writes in the topic's data stream, tracks aborted transaction IDs, and gates the dispatcher's read horizon (`maxReadPosition`) so that uncommitted entries are not delivered. The TB persists its state in a per-namespace system topic (`__transaction_buffer_snapshot`). +- **PendingAckStore** — a per-(topic, subscription) component that records transactional acknowledgments in a sibling persistent topic (`-__transaction_pending_ack`), applying them to the cursor only when the transaction commits. + +When a transaction ends, the TC sends `END_TXN_ON_PARTITION` (and `END_TXN_ON_SUBSCRIPTION` for acks) to every participant. The TB then writes a **commit or abort marker** as a regular entry in the topic's managed ledger. The dispatcher discovers committed/aborted state by replaying these markers and consulting the in-memory aborted-txn set. + +### Scalable topics + +[PIP-460](pip-460.md) introduces scalable topics: a logical topic backed by a DAG of range segments (`segment://...`) that can be split or merged at runtime. Each segment is a regular `PersistentTopic` from the broker's perspective, but the segment's lifetime is controlled by the [scalable topic controller](pip-468.md) — segments get **sealed** when split or merged, after which the segment's managed ledger no longer accepts writes. + +### How the two interact + +The current transaction implementation composes per-`PersistentTopic`. With scalable topics, every segment carries its own TB. This composition fails in two ways: + +1. **End-of-transaction stalls on sealed segments.** The TC sends `END_TXN_ON_PARTITION` to each segment that received writes. The segment's TB tries to append a commit/abort marker — which is a write — and the now-sealed segment rejects it. The end-txn RPC times out (~30s). +2. **Pending-ack topic naming collides with the segment-domain parser.** The convention `-__transaction_pending_ack` is unparseable when `` is a `segment://...` URI. (Worked around in #25631 with a flat persistent name; see "Out of Scope" below.) + +The first issue is structural, not just a routing bug. As long as commit/abort decisions need to be persisted **inside the topic's data stream**, sealing the topic terminates any in-flight transaction. + +--- + +## Motivation + +We need transactions that: + +1. Provide atomicity across multiple writes and acknowledgments, possibly spanning multiple topics across multiple namespaces. +2. Compose correctly with the scalable-topic lifecycle — including splits, merges, and segments sealed mid-transaction. +3. Do not require duplicating data (each `producer.send` produces a single managed-ledger append). +4. Reuse as much of the existing transaction surface as possible — interfaces, dispatcher integration, client API — so that we are not re-litigating well-understood concerns. +5. Coexist with v4 transactions on `persistent://` topics with no behavior change for those topics. + +The structural mismatch between in-stream markers and a mutable segment DAG cannot be papered over at the routing or the topic-naming layer. It needs a transaction representation that does not put the decision record inside the data stream. + +--- + +## Goals + +### In Scope + +- Atomic transactions over `segment://` topics (writes and acks), including transactions whose lifetime spans split/merge. +- Multi-topic, multi-namespace, multi-segment transactions with the same atomicity guarantees as today. +- Reuse of the existing `Transaction`, `TransactionCoordinator`, `TransactionBuffer`, `PendingAckStore`, dispatcher, and client APIs. New behavior arrives as alternative implementations behind the existing interfaces. +- Coexistence with the legacy in-stream-marker implementation for `persistent://` topics. + +### Out of Scope + +- Replacing the legacy implementation for non-scalable topics. The new implementation is opt-in per topic; `persistent://` topics keep their current behavior, including the existing TC. +- Replacing the segment-aware pending-ack topic name introduced in #25631 — that workaround becomes unnecessary as a side effect of this PIP and is removed in the same change. +- Cross-cluster (geo-replicated) transactional semantics. + +--- + +## High Level Design + +The proposal is one sentence: + +> **Move transactional state out of the data stream and into the metadata store.** + +Concretely: keep all existing components and interfaces, and add a parallel implementation of `TransactionBuffer`, `PendingAckStore`, **and Transaction Coordinator** that writes nothing to any data stream. Their state lives entirely in the metadata store. The legacy in-stream-marker components remain, unchanged, for `persistent://` topics; the new metadata-driven components handle `segment://` topics. The dispatcher's contract is unchanged. + +Why introduce a v5 TC rather than reuse the legacy one: the legacy TC stores its log in a system topic (`__transaction_log_*`), which carries the operational concerns of any system topic — compaction can lead to long recovery times, leadership has to be maintained, and recovery is on the data path. With the metadata store available we can have a TC whose state is just a few key-value records, no log, no system topic, no per-broker in-memory replay. Running both TC implementations in parallel keeps v4 transactions byte-for-byte unchanged while the v5 path uses the simpler design. + +### Why this works for scalable topics + +- **Sealing a segment is irrelevant.** Commit/abort no longer require any append to the segment. End-txn becomes a metadata-store CAS on a single record. Sealed segments materialize the decision (advance cursors, evict cache entries) without writing anything. +- **The dispatcher does not change.** It already asks the topic's TB for `maxReadPosition` and `isTxnAborted`. We swap the source. +- **Splits/merges do not strand transactions.** Sealed parents and live children both consult the same metadata; the decision lives above the segments. +- **No data is duplicated.** Each transactional `send` produces exactly one managed-ledger append, same as today. + +### Architecture Overview + +``` +┌──────────────────────────────────────────────────────────────────┐ +│ Client (V5) -- producer.send(txn,...) │ +│ -- consumer.acknowledge(id, txn) │ +└─────────────────────────────────┬─────────────────────────────────┘ + │ + ┌─────────────────────┴─────────────────────┐ + │ │ +┌───────────▼────────────────┐ ┌────────────────▼──────────────┐ +│ Transaction Coordinator │ │ Transaction Coordinator V5 │ +│ (legacy, BK-backed log) │ │ (metadata-store records) │ +│ → v4 / persistent:// txns │ │ → v5 / segment:// txns │ +└───────────┬────────────────┘ └────────────────┬──────────────┘ + │ │ + │ END_TXN_ON_PARTITION / SUBSCRIPTION │ + ▼ ▼ +┌─────────────────────────────────────────────────────────────────────┐ +│ Per-topic broker components │ +│ │ +│ ┌──────────────────────────┐ ┌────────────────────────┐ │ +│ │ TopicTransactionBuffer │ │ MLPendingAckStore │ │ +│ │ (in-stream markers) │ │ (sibling topic) │ │ +│ │ → persistent:// topics │ │ → persistent:// topics│ │ +│ └──────────────────────────┘ └────────────────────────┘ │ +│ │ +│ ┌──────────────────────────┐ ┌────────────────────────┐ │ +│ │ MetadataTransactionBuffer │ │ MetadataPendingAckStore│ │ +│ │ (metadata-store records) │ │ (metadata-store records) │ +│ │ → segment:// topics │ │ → segment:// topics │ │ +│ └────────┬─────────────────┘ └─────────┬──────────────┘ │ +└────────────┼──────────────────────────────┼─────────────────────────┘ + │ │ + ▼ ▼ + Metadata Store — txn coordinator state + txn-op records + secondary indexes +``` + +The `TransactionBufferProvider` and `TransactionPendingAckStoreProvider` SPIs already exist. The new TB / PendingAckStore implementations slot in behind them. The v5 TC is a parallel coordinator selected by the client when it is configured for the new path. Selection on the participant side is per-topic, based on the topic's domain. + +--- + +## Detailed Design + +### Data Model + +The metadata store holds two classes of records and four secondary indexes. All records for a given transaction share the same **partition key** (`txnId`) so they are co-located — this makes per-txn scans (e.g. listing all ops to apply at end-txn time) a single-partition operation rather than a fan-out. + +> **A note on metadata-store backends.** The design is `MetadataStore`-agnostic. It depends on three capabilities — partition-key co-location, sequential keys, and secondary indexes with range queries and range-watch — that the `MetadataStore` interface does not expose today. We extend the interface to surface them; backends that natively support these (notably Oxia, the intended default) implement them directly, while backends that don't (e.g. ZooKeeper) can implement them in a less efficient way (client-side counters for sequential IDs; client-maintained index records; periodic re-list in lieu of range-watch). Correctness does not depend on backend choice; throughput and recovery latency may. + +#### Header — one per transaction. Linearization point. + +``` +/txn/ partitionKey = txnId + = { + state: OPEN | COMMITTED | ABORTED, + timeout_ms: , + created_ms: + } +``` + +State transitions are conditional puts (CAS on version) issued by the v5 TC. `OPEN → COMMITTED` and `OPEN → ABORTED` are the only allowed transitions; `COMMITTED` and `ABORTED` are terminal. + +#### Operation records — one per transactional write or ack. Unbounded. + +``` +/txn-op// partitionKey = txnId, + sequential = true # server-assigned + = { + kind: "write" | "ack", + segment: "segment://t/n/x/", # always present + subscription: "", # ack only + position: : + } +``` + +Each operation is its own record, so a transaction has no size limit and concurrent participants do not contend on a single record. With **sequential keys** the server (or, on backends that lack them, a `MetadataStore`-side counter) assigns ``, eliminating client-side collisions. + +#### Secondary indexes (auto-maintained by the metadata store) + +``` +idx:writes-by-segment on /txn-op/* where kind=write + key = segment + → range query "writes touching segment S" + +idx:acks-by-segment-subscription on /txn-op/* where kind=ack + key = (segment, subscription) + → range query "acks on (segment S, subscription SU)" + +idx:txn-by-deadline on /txn/* where state=OPEN + key = timeout_ms + → range query "open txns past deadline" + → used by TC for timeout-driven abort + +idx:txn-by-final-state on /txn/* where state ∈ {COMMITTED, ABORTED} + key = (state, finalized_ms) + → range query "finalized txns ready for GC" + → used by GC sweep to find finalized txns whose op records can be deleted +``` + +#### Garbage collection + +A finalized transaction (`COMMITTED` or `ABORTED`) is removed in two phases: + +1. **Per-participant materialization.** When the TC fans out end-txn, each participant broker materializes the decision (commit: advance subscription cursors for acks, evict header cache; abort: drop ops). Once a participant has finished its materialization for ``, it deletes its op records (`/txn-op//` for ops it owns). +2. **Header GC sweep.** A periodic sweep scans `idx:txn-by-final-state` for entries past a configurable retention window (e.g. 60 s after `finalized_ms`). For each, it verifies no `/txn-op//*` records remain (orphan check from a participant crash), forces deletion of any leftovers, and finally deletes the header `/txn/`. + +Because all of a txn's records share the same partition (`partitionKey = txnId`), the GC sweep's per-txn cleanup stays in one partition: list `/txn-op//`, delete, then delete the header. + +Indexes update transactionally with the underlying records, so they self-clean. + +### Components + +#### `MetadataTransactionBuffer` (new) + +Implements the existing `TransactionBuffer` interface. Used for `segment://` topics. + +| Method | Behavior | +|---|---| +| `appendBufferToTxn(txnId, buf)` | `ML.asyncAddEntry(buf)`; on success, append a sequential `/txn-op//` (`partitionKey=txnId`) with `kind="write", segment, position`. The publish ack waits for both. | +| `commit(txnId, position)` / `abort(...)` | Not invoked by the v5 TC (which does not RPC participants). The TB's header watch fires when `/txn/.state` changes; the TB then materializes locally (evict / mark-aborted) and deletes its owned op records. | +| `getMaxReadPosition()` | Read from in-memory cache. Cache is populated by a watch on `idx:writes-by-segment == ` joined against the header cache. Result: `min(position over OPEN txns) - 1`, capped at LAC. | +| `isTxnAborted(msg)` | Look up `/txn/.state` from header cache. | +| `recover()` | Open the index watch and the header cache; populate from the current snapshot. No log replay, no snapshot topic. | + +#### `MetadataPendingAckStore` (new) + +Implements the existing `PendingAckStore` interface. Used for `segment://` topic subscriptions. + +| Method | Behavior | +|---|---| +| `appendIndividualAck(txnId, positions)` | Append sequential `/txn-op//` records with `kind="ack", segment, subscription, position`. | +| `appendCumulativeAck(...)` | Same shape, single op record carrying the cumulative position. | +| `commit(txnId)` / `abort(txnId)` | Not invoked by the v5 TC. Triggered locally when the header watch on `/txn/.state` fires. Commit: range-query `idx:acks-by-segment-subscription == (, )` filtered to ``; apply to cursor (`markDelete` or `individualAck`); range-delete the op records. Abort: range-delete the op records, no cursor work. | +| `replayAsync()` (recovery) | Range-query `idx:acks-by-segment-subscription == (, )`, group by `txnId`, hydrate in-memory state. | + +#### Transaction Coordinator V5 (new) + +A parallel coordinator selected by the v5 client. Same client-facing wire commands (`NEW_TXN`, `ADD_PARTITION_TO_TXN`, `ADD_SUBSCRIPTION_TO_TXN`, `END_TXN`), but no system-topic log: every operation reads or CAS's a metadata-store record. **The TC does not RPC participants** — see "Notification mechanism" below. + +| Operation | Behavior | +|---|---| +| `newTxn(timeoutMs)` | Create `/txn/` with `state=OPEN`, `timeout_ms=now+timeoutMs`. | +| `addPartitionToTxn` / `addSubscriptionToTxn` | No-op at the coordinator. The participant broker writes its own op records when the actual write/ack arrives; the TC never needs to enumerate participants. | +| `endTxn(COMMIT\|ABORT)` | A single CAS on `/txn/.state`. After it returns, the TC sets `finalized_ms` on the header and acks the client. No fan-out, no waiting on participants. | +| Timeout sweep | Range-query `idx:txn-by-deadline` for entries with `timeout_ms ≤ now`, abort each (same single-CAS flow). | +| GC sweep | Range-query `idx:txn-by-final-state` for entries past retention; for each, verify `/txn-op//*` is empty (force-delete leftovers); delete header. | + +Why parallel rather than reusing the legacy TC: the legacy TC's per-shard system topic (`__transaction_log_*`) requires leadership election, runs compaction over its own log, and pays a recovery cost on every broker restart proportional to the live transaction set. The v5 TC's state is just per-txn KV records — there is no log to compact and no cold-start replay. Running both in parallel keeps v4 transactions byte-for-byte unchanged while the v5 path uses the simpler design. A v5 client routes its `NEW_TXN` to the v5 TC; v4 clients route to the legacy TC. A single transaction does not span the two. + +#### Notification mechanism (TC → participants) + +The legacy TC needs to RPC each participant (`END_TXN_ON_PARTITION`, `END_TXN_ON_SUBSCRIPTION`) because the participants have no other way to learn the decision — the TC's log is the only source of truth, and only the TC reads it. + +In the v5 design **the metadata store is the source of truth**, and every participant already reads from it. Participants therefore learn about state transitions directly from the store, without any TC-to-broker RPC: + +- A `MetadataTransactionBuffer` keeps an in-memory header cache for txns it has writes from. The cache entries are populated when a write op record is appended (the broker reads the header to authorize the write) and **kept up to date by point-watches on the headers it has cached**. +- A `MetadataPendingAckStore` maintains the same pattern for txns it has acks from. +- When the TC CAS's `/txn/.state` from OPEN → COMMITTED/ABORTED, every cached watcher fires. Each participant materializes locally: + - **Commit** — TB evicts its cache entry (the txn no longer pins `maxReadPosition` back); PendingAckStore applies the buffered acks to the cursor. + - **Abort** — TB marks the txn aborted in its cache (the dispatcher's `isTxnAborted` will skip those entries); PendingAckStore drops the buffered acks. +- After materialization, the participant deletes the op records it owns (`/txn-op//` for ops on its segment / subscription). +- The TC's GC sweep (above) detects when all participants have done their cleanup — the prefix `/txn-op//*` is empty — and deletes the header. + +Consequences: + +- **End-txn latency.** From the client's perspective, `commit` returns as soon as the header CAS lands. From a consumer's perspective, freshly-committed entries become visible after the participant's header watch fires + materialization runs. That's typically tens of milliseconds; bounded by metadata-store watch propagation. (If we ever care about a tighter bound — e.g. for a given workload — the TC can issue an optional `nudge` RPC to participants in parallel with the CAS. Not needed for correctness; not in this PIP.) +- **No RPC fan-out from TC.** End-txn is `O(1)` work at the TC: one CAS. The cost of fan-out is paid by the metadata store's watch-delivery infrastructure, which already exists for other Pulsar uses. +- **Crash idempotence.** A participant that crashes during materialization restarts, observes the (already-final) header state via its watch, and finishes materialization. The TC need not retry anything. + +#### Dispatcher + +Unchanged. It already asks `topic.getTransactionBuffer().getMaxReadPosition()` and `topic.getTransactionBuffer().isTxnAborted(...)`. The new TB implements both. + +### Flows + +#### Publish (transactional) + +```mermaid +sequenceDiagram + participant C as Client + participant B as Segment broker + participant ML as Managed Ledger + participant M as Metadata Store + + C->>B: send(txnId, payload) + B->>M: read /txn/.state (cached) + alt state != OPEN + B-->>C: TxnConflict + else state == OPEN + B->>ML: asyncAddEntry(payload) + ML-->>B: position + B->>M: put /txn-op// {kind=write, segment, position} + M-->>B: ack + B-->>C: send-ack + end +``` + +The header read is cache-first; the cache is invalidated by the same watch the TB already maintains on the header. The op-record put is the only synchronous metadata-store write on the publish path. + +#### End-txn (commit or abort) + +```mermaid +sequenceDiagram + participant Cl as Client + participant TC as Transaction Coordinator V5 + participant M as Metadata Store + participant P as Participant brokers + + Cl->>TC: commit(txnId) + TC->>M: CAS /txn/.state OPEN→COMMITTED, set finalized_ms + M-->>TC: ack + TC-->>Cl: ack + + Note over M,P: Independently, asynchronously: + M-->>P: header watch fires + P->>P: materialize (cursor advance / cache evict) + P->>M: delete owned /txn-op// records + + Note over TC,M: Later, GC sweep: + TC->>M: list /txn-op//* (empty? then delete header) +``` + +The CAS on the header is the linearization point — that is when the transaction's outcome is decided. Notification of participants is not part of the linearization; it propagates via the watches every participant already maintains on the headers it has cached. Sealed segments are fine — materialization is metadata + cursor work, no managed-ledger writes. + +#### Subscribe / dispatch + +Unchanged. The dispatcher polls `tb.getMaxReadPosition()` and filters by `tb.isTxnAborted(msg)`. The `MetadataTransactionBuffer` answers both from its in-memory caches, fed by metadata-store watches. + +#### Late-write race + +The TC is mid-end-txn when the client publishes once more inside the same transaction. The header CAS may have already flipped to `COMMITTED`/`ABORTED`. The publish-path header check on the participant broker rejects with `TxnConflict`. This mirrors today's TC behavior (the TC marks transactions as ENDING and brokers reject new writes); the only difference is that the rejection criterion is now read from the metadata store rather than from a TC RPC. + +### Recovery + +- **Broker startup.** Each `MetadataTransactionBuffer` opens its index watch and header cache. The first watch event delivers the snapshot; the TB is ready as soon as the snapshot has been applied. No log replay, no system-topic reader, no snapshot topic. +- **Broker crash mid-publish.** If the broker appended the entry but crashed before writing the op record, the entry exists in the segment but no metadata claims it. On txn timeout the TC aborts the txn; the dispatcher's `isTxnAborted` check (which falls back to "abort" for unknown txnIds at retention horizon) discards the entry. +- **Broker crash mid-end-txn.** If the header CAS landed but materialization on a participant did not complete, the participant re-derives state from the header on restart and finishes materialization. End-txn is idempotent. +- **TC failover.** The v5 TC has no in-memory log to replay — its state lives in the metadata store. Whichever broker takes over coordinator duty for a TC partition resumes operations directly from the metadata-store records. Cold-start cost is bounded by `idx:txn-by-deadline` and `idx:txn-by-final-state` scans, not by replay of an entire transaction log. + +### Concurrency and contention + +- Each transactional publish writes a unique `/txn-op//` record (server-assigned sequential key). There is no contention between concurrent participants of the same transaction. +- The header is CAS'd at most twice per transaction lifetime (open + finalize), so contention there is bounded. +- All records for a given txn share `partitionKey=txnId`, so per-txn operations (list, range-delete) stay on a single partition. +- Index updates are managed by the metadata store; their scaling is the store's concern. + +--- + +## Public-facing Changes + +### Public API + +No changes. The client-facing `Transaction` API is unchanged. + +### Binary protocol + +No changes to client-facing wire commands (`NEW_TXN`, `ADD_PARTITION_TO_TXN`, `ADD_SUBSCRIPTION_TO_TXN`, `END_TXN`) — the v5 TC accepts them with the same semantics as the legacy TC. + +The broker-to-broker commands `END_TXN_ON_PARTITION` and `END_TXN_ON_SUBSCRIPTION` are **not used** by the v5 path: participant brokers learn about the decision by watching the metadata-store header rather than by receiving an RPC from the TC. The legacy TC still uses these commands for v4 transactions; they remain in the protocol unchanged. + +### Configuration + +A per-namespace or per-broker setting selects the TB implementation. Default for `segment://` topics: metadata-driven. Default for `persistent://` topics: in-stream markers (unchanged). Override is possible per-namespace for debugging / migration. + +### Metrics + +Existing transaction metrics remain. The metadata-driven implementation adds: + +- `pulsar_txn_metadata_store_op_writes_total` (counter) — op records written. +- `pulsar_txn_metadata_store_header_cas_total{result="ok|conflict|reject"}` (counter) — header CAS attempts and outcomes. +- `pulsar_txn_metadata_store_index_query_seconds` (histogram) — latency of the index range queries on `idx:writes-by-segment` / `idx:acks-by-segment-subscription`. +- `pulsar_txn_metadata_store_outstanding_op_records` (gauge) — uncollected op records (a proxy for txn GC backlog). + +Existing `pulsar_txn_tb_*` snapshot/replay metrics are not emitted by the new implementation (no snapshots, no replay). + +--- + +## Backward & Forward Compatibility + +### Upgrade + +- Existing `persistent://` topic behavior is unchanged. v4 clients see no difference. +- Brokers running this PIP can interoperate with brokers that do not, as long as a given **topic** is consistently served by brokers of one kind. Since topic ownership is bundle-based and migration via load-balancer transfers TB state across, this is satisfied automatically. +- Per-segment pending-ack topics created by the workaround in #25631 (`persistent://t/n/--__transaction_pending_ack`) are no longer used. They are deleted as part of upgrade. Since the workaround was only ever exercised by V5 transactional consumer flows, the upgrade path is safe. + +### Downgrade / Rollback + +Not applicable. Scalable topics are introduced as a new feature in Pulsar 5.0 ([PIP-460](pip-460.md)); this PIP defines transactional support for that feature from the start. There is no prior version to roll back to. + +### Pulsar Geo-Replication + +Out of scope. Transactional geo-replication is not supported in either model. + +--- + +## Alternatives Considered + +### A. Move TB to the scalable-topic level (one TB per logical topic) + +Earlier draft of this design. Architecturally clean — decisions live above segments — but introduces a new broker-side singleton per scalable topic, adds new failover semantics, and complicates the TC's wire protocol (end-txn would need redirection from segment to scalable-topic owner). Replacing the per-topic TB **implementation** with a metadata-driven one achieves the same correctness without any of that surface area. + +### B. Per-segment TB but using an off-segment marker stream + +Keep the per-segment TB; have it write commit/abort markers to a **separate** managed ledger (e.g. a shadow topic) rather than into the segment's own data. Sealed segments would no longer block end-txn. Rejected because: (1) it doubles the data path (every txn needs a write to the segment **and** to the shadow topic), (2) it requires a new system-topic-per-segment, and (3) it does not eliminate the snapshot/replay machinery that the metadata-driven approach removes outright. + +### C. Skip transactional support on scalable topics + +Document scalable topics as non-transactional. Rejected: the transactional consume-and-produce pattern is a primary use case for scalable streaming workloads (Kafka Streams analogue), and PIP-460's roadmap explicitly calls out transactions across range segments as a Phase 4 deliverable. + +--- + +## General Notes + +The shape of the change is *one new `TransactionBuffer` implementation, one new `PendingAckStore` implementation, one new `TransactionCoordinator` implementation, and the `MetadataStore` extensions to support partition-key co-location, sequential keys, and secondary indexes with range-watch*. The complexity is in the interaction of the metadata schema with the dispatcher's existing assumptions, not in any new system component on the broker. + +## Links + +- [PIP-460: Scalable Topics](pip-460.md) +- [PIP-468: Scalable Topic Controller](pip-468.md) +- Mailing List discussion thread: TBD +- Mailing List voting thread: TBD From d61695f8e9a34969b6a1cf7c56aeb966fbd746e9 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 6 May 2026 12:39:32 -0700 Subject: [PATCH 2/3] Update pip/pip-471.md Co-authored-by: Lari Hotari --- pip/pip-471.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pip/pip-471.md b/pip/pip-471.md index 2ef59c0c27a15..2c4583178b7a6 100644 --- a/pip/pip-471.md +++ b/pip/pip-471.md @@ -1,4 +1,4 @@ -# PIP-471: Metadata-Driven Transactions for Scalable Topics +# PIP-473: Metadata-Driven Transactions for Scalable Topics *Sub-PIP of [PIP-460: Scalable Topics](pip-460.md)* From 2f41af480e38c0676aae4a517a0f954c194ba54b Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 6 May 2026 12:39:58 -0700 Subject: [PATCH 3/3] Rename pip-471.md to pip-473.md --- pip/{pip-471.md => pip-473.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename pip/{pip-471.md => pip-473.md} (100%) diff --git a/pip/pip-471.md b/pip/pip-473.md similarity index 100% rename from pip/pip-471.md rename to pip/pip-473.md