[improve] Add Option.SequenceKeysDeltas + subscribeSequence#25724
Merged
Conversation
… with Oxia native + compat layer
Add a new Option subtype, Option.SequenceKeysDeltas(List<Long> deltas), that
when present on put(...) requests server-assigned multi-dimensional atomic
sequence-key suffixes — same semantics and key format as Oxia's native
sequence-keys. Add a companion subscription method, MetadataStore.subscribeSequence,
that delivers the latest assigned sequence key to a listener as new sequence
records appear under the prefix.
Oxia: native passthrough. doStorePut translates SequenceKeysDeltas to
PutOption.SequenceKeysDeltas; subscribeSequence delegates to
client.getSequenceUpdates. supportsNativeSequenceKeys() returns true.
Compatibility layer in AbstractMetadataStore: when the backend has no native
sequence-keys, intercept put(prefix, ..., {SequenceKeysDeltas}) at the top of
put(), CAS-increment a sidecar counter document at <prefix>__seq_counter__,
synthesize the actual key as <prefix>-<seq:%020d>-... matching Oxia byte-for-byte,
then recurse into the regular put path with the synthesized key. subscribeSequence
filters Created notifications by prefix and emits monotonically.
Wrappers: DualMetadataStore + FaultInjectionMetadataStore forward subscribeSequence
to the wrapped store.
Tests:
- OxiaSequenceKeysTest: 3-shard Oxia cluster — native single/multi-dim + subscribe
- SequenceKeysTest: cross-backend (ZK/Memory/RocksDB/Oxia/MockZooKeeper) — same
three scenarios. Both layers must produce monotonically increasing keys with
matching byte-format and deliver subscription updates.
merlimat
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
Follow-up to #25723. Add atomic, multi-dimensional sequence keys to the
MetadataStoreAPI — same model as Oxia's native sequence-keys — and a companion subscription channel for change notifications. PIP-471 (transaction state via metadata) needs both: per-segment append-only logs (/txn-op/<txnId>-<seq>) where the server picks the suffix atomically, and a notification stream so each TB can apply only its own segment's events.The existing
Option.Sequential(single-dim, +1 only, two-write fallback on Oxia) is preserved; this PR adds a richer primitive alongside it.Modifications
API:
Option.SequenceKeysDeltas(List<Long> deltas)— when present onput(prefix, ...), the actual stored key becomes<prefix>-<seq0:%020d>-<seq1:%020d>-.... Each dimension increments atomically by its delta (first must be > 0, rest >= 0). The returnedStatcarries the actual generated path.MetadataStore.subscribeSequence(prefix, listener, opts) → AutoCloseable— listener receives the latest assigned sequence key underprefixas new records appear. Multiple updates may collapse into a single emission with the highest sequence. Closing the handle unsubscribes.OptionsHelper.sequenceKeysDeltas(Set<Option>)accessor.Oxia (native):
OxiaMetadataStoretranslatesOption.SequenceKeysDeltastoPutOption.SequenceKeysDeltas;subscribeSequencedelegates toclient.getSequenceUpdates. MarkssupportsNativeSequenceKeys() == trueso the compat layer is bypassed.Compatibility layer in
AbstractMetadataStore(used by LocalMemory, RocksDB, ZooKeeper, MockZooKeeper):putinterceptsOption.SequenceKeysDeltas, runs a CAS-retry loop on a sidecar counter document at<prefix>__seq_counter__(a fixed-width binary blob oflongs), then recurses intoputwith the synthesized key. Synthesized keys match Oxia's byte format exactly.subscribeSequenceregisters an internalNotificationlistener that filtersCreatedevents whose paths start with<prefix>-, tracks the highest path seen via CAS, and delivers monotonically. Closing the handle removes the listener.Wrappers:
DualMetadataStoreandFaultInjectionMetadataStoreforwardsubscribeSequenceto the wrapped store.Verifying this change
Two new test classes:
OxiaSequenceKeysTest— 3-shard Oxia container, exercises the native path: single/multi-dim sequence puts, subscribe collapses to highest emission.SequenceKeysTest— runs on everyBaseMetadataStoreTestbackend (ZK, Memory, RocksDB, Oxia, MockZooKeeper). Both compat and native paths must produce monotonically increasing keys with matching byte-format and deliver subscription updates.Local results: 15/15 cross-backend tests pass; full metadata test suite (
./gradlew :pulsar-metadata:test) green.Does this pull request potentially affect one of the following parts:
MetadataStoreAPI — strictly additive: a newOptionsubtype and a new default method onMetadataStore. Existing callers and external implementations are unaffected.Matching PR in forked repository
PR in forked repository: https://github.com/merlimat/pulsar/pull/new/mmerli/metadata-sequence-keys