lhotari commented on code in PR #25693:
URL: https://github.com/apache/pulsar/pull/25693#discussion_r3197112755
##########
pip/pip-473.md:
##########
@@ -0,0 +1,401 @@
+# PIP-473: 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
(`<topic>-<sub>__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 `<topic>-<sub>__transaction_pending_ack` is unparseable when
`<topic>` 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/<txnId> partitionKey = txnId
+ = {
+ state: OPEN | COMMITTED | ABORTED,
+ timeout_ms: <abs epoch ms>,
+ created_ms: <abs epoch 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/<txnId>/<seq> partitionKey = txnId,
+ sequential = true # server-assigned
<seq>
+ = {
+ kind: "write" | "ack",
+ segment: "segment://t/n/x/<descriptor>", # always present
+ subscription: "<sub-fqn>", # ack only
+ position: <ledgerId>:<entryId>
+ }
+```
+
+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 `<seq>`, 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 `<txnId>`, it deletes its op records
(`/txn-op/<txnId>/<seq>` 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/<txnId>/*` records remain (orphan check from
a participant crash), forces deletion of any leftovers, and finally deletes the
header `/txn/<txnId>`.
Review Comment:
Local Claude Code review comment (I'm not sure how accurate this is):
> [BUG] GC orphan-cleanup window vs. slow participant can lose committed
work. Step 1 (per-participant materialization) and step 2 (header GC sweep)
interleave optimistically. If a participant is partitioned/slow and hasn't
materialized when the 60s retention window elapses, the GC sweep finds the
participant's /txn-op/<txnId>/* records still present and "forces deletion of
any leftovers". When the participant reconnects and runs
MetadataPendingAckStore.commit for that txn, its
idx:acks-by-segment-subscription range query returns empty, so no acks are ever
applied to the cursor — silent loss of committed acks. The text needs either a
much larger retention window with rationale, a participant-liveness check
before forced deletion, or watcher-driven materialization that is acknowledged
before the GC may proceed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
