[feat] PIP-473: Metadata-Driven Transactions for Scalable Topics by merlimat · Pull Request #25693 · apache/pulsar

merlimat · 2026-05-06T00:24:58Z

Summary

Proposes a transaction model for scalable topics that replaces the in-stream commit/abort marker mechanism with metadata-store-backed state, so transactions compose correctly with the segment lifecycle (splits, merges, sealed segments).

The shape of the change:

MetadataTransactionBuffer — new TransactionBuffer implementation for segment:// topics. Writes nothing to the topic's data stream; transactional state lives as records and secondary indexes in the metadata store.
MetadataPendingAckStore — new PendingAckStore implementation for segment:// subscriptions. No sibling pending-ack topic.
Transaction Coordinator V5 — parallel coordinator selected by v5 clients. Single CAS per end-txn; no system-topic log, no log compaction, no cold-start replay. Participants observe state transitions via metadata-store header watches — no broker-to-broker RPCs.
MetadataStore extensions — surface partition-key co-location, sequential keys, and secondary indexes with range-watch. Default backend (Oxia) maps to native primitives; ZooKeeper and others can implement them sub-optimally without affecting correctness.

Existing persistent:// topic behavior is unchanged. v4 transactions keep their current implementation byte-for-byte.

The PIP is self-contained — full data model, end-to-end flows (publish, end-txn, dispatch, recovery, GC), backward-compat, and rejected alternatives are in the document.

Test plan

Mailing list discussion
Mailing list vote

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.

Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>

lhotari

I've added some review comments and questions.

lhotari · 2026-05-06T18:25:37Z

+- **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.


regular entry -> marker entry

lhotari · 2026-05-06T18:28:45Z

+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.


Does this mean a transaction ID for a v4 or v5 transaction can be used interchangeably between the two clients?

lhotari · 2026-05-06T18:32:20Z

+> **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.


What would be the migration solution and possible rollback solution for existing v4 topic data which has been written in v4 transactions format?

lhotari · 2026-05-06T18:38:14Z

+
+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).


is it possible to remove any op records for finalized txns until the data has been deleted? as long as the data is retained, do the txn records need to exist so that the entries can be consistently consumed? Without txn commit/abort information, it wouldn't be possible to determine whether certain records should be skipped or not when a new subscription or reader/checkpointconsumer reads the records.

lhotari · 2026-05-06T19:50:50Z

+
+#### 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.


There's currently a performance issue (either additional latency or CPU spinloop) when dispatcher polls getMaxReadPosition. This happens when the dispatcher has read all entries up to the maxReadPosition.

For example for the read operation in PersistentDispatcherMultipleConsumers.readMoreEntries:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

Lines 420 to 421 in 138595f

cursor.asyncReadEntriesWithSkipOrWait(messagesToRead, bytesToRead, this, ReadType.Normal,

topic.getMaxReadPosition(), createReadEntriesSkipConditionForNormalRead());

The read will complete here:

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

Lines 2314 to 2317 in a39d241

if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {

opReadEntry.checkReadCompletion();

return;

}

After completion, a new read will be issued. This loop will keep on going until the maxReadPosition moves forward.

A better solution would be that the dispatcher could be notified by the transaction component when the maxReadPosition has advanced so that there wouldn't be a need to poll.

btw. The current dispatcher code should be optimized for v4 transactions since even with polling, it could be improved.

lhotari · 2026-05-06T20:00:03Z

+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>`.


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//* 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.

github-actions Bot added the PIP label May 6, 2026

merlimat changed the title ~~[PIP-471] Metadata-Driven Transactions for Scalable Topics~~ [feat] PIP-471: Metadata-Driven Transactions for Scalable Topics May 6, 2026

lhotari changed the title ~~[feat] PIP-471: Metadata-Driven Transactions for Scalable Topics~~ [feat] PIP-473: Metadata-Driven Transactions for Scalable Topics May 6, 2026

lhotari reviewed May 6, 2026

View reviewed changes

Comment thread pip/pip-471.md Outdated

merlimat and others added 2 commits May 6, 2026 12:39

Update pip/pip-471.md

d61695f

Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>

Rename pip-471.md to pip-473.md

2f41af4

lhotari reviewed May 6, 2026

View reviewed changes

merlimat mentioned this pull request May 6, 2026

[improve][metadata] Add streaming scanChildren to MetadataStore #25701

Open

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[feat] PIP-473: Metadata-Driven Transactions for Scalable Topics#25693

[feat] PIP-473: Metadata-Driven Transactions for Scalable Topics#25693
merlimat wants to merge 3 commits intoapache:masterfrom
merlimat:pip-471-metadata-driven-transactions

merlimat commented May 6, 2026

Uh oh!

Uh oh!

lhotari left a comment

Uh oh!

lhotari May 6, 2026

Uh oh!

lhotari May 6, 2026

Uh oh!

lhotari May 6, 2026

Uh oh!

lhotari May 6, 2026

Uh oh!

lhotari May 6, 2026

Uh oh!

lhotari May 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants


		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).


		#### 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.

	cursor.asyncReadEntriesWithSkipOrWait(messagesToRead, bytesToRead, this, ReadType.Normal,
	topic.getMaxReadPosition(), createReadEntriesSkipConditionForNormalRead());

	if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {
	opReadEntry.checkReadCompletion();
	return;
	}

		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>`.

Conversation

merlimat commented May 6, 2026

Summary

Test plan

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

lhotari May 6, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari May 6, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari May 6, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari May 6, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari May 6, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari May 6, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants