Scalable Topics: pulsar::st producer — per-segment fan-out, key routing, MessageId#601
Open
merlimat wants to merge 8 commits into
Open
Scalable Topics: pulsar::st producer — per-segment fan-out, key routing, MessageId#601merlimat wants to merge 8 commits into
pulsar::st producer — per-segment fan-out, key routing, MessageId#601merlimat wants to merge 8 commits into
Conversation
The per-segment producers of a scalable topic attach to its segment:// backing topics through the classic producer machinery, so the classic TopicName must understand the domain - while the classic user-facing API must keep rejecting it (only the pulsar::st client may touch segment topics). Ported from the Java client's TopicName/PulsarClientImpl changes. - TopicName: new TopicDomain::Segment. segment://<tenant>/<ns>/<parent>/ <descriptor> always parses as the new (cluster-less) format - the local name contains a '/', so the token-count heuristic must not mistake it for a legacy V1 name - and requires the descriptor component. isPersistent() is true for segment topics (they are the persistent backing topics of one segment, matching Java); new isSegment() accessor. Classic persistent/non-persistent parsing is unchanged, and topic:// (the scalable-topic identity scheme) intentionally stays unparseable here - the st client resolves it before this layer. - ClientImpl: every user-facing entry point (createProducerAsync, the subscribe funnel, createReaderAsyncV2, createTableViewAsyncV2, getPartitionsForTopicAsync) now rejects segment topics with ResultInvalidTopicName and a message pointing at the pulsar::st API, mirroring the Java client's scalable-domain check. The new createSegmentProducerAsync() bypasses the rejection for lib/st (the Java-parity escape hatch); consumer-side bypasses come with their phases. - tests/st/StTopicNameTest.cc: segment parse (fields, slashed local name, roundtrip, persistent/segment flags, never partitioned), descriptor requirement, classic-format regression, and broker-free rejection of segment topics through the classic producer/subscribe/reader API. Verified: 74/74 pulsar-st-tests green; the classic TopicNameTest (16 tests, broker-free) and pulsar-tests/pulsarShared builds stay green; clang-format-11 clean. Signed-off-by: Matteo Merli <mmerli@apache.org>
The scalable-topic MessageId, ported field-for-field from the Java client's
MessageIdV5 so serialized ids interoperate between the two clients.
- MessageIdImpl wraps the classic per-segment MessageId plus the segment it
belongs to (segmentId -1 = the cross-segment sentinels), with the optional
sections the consumer side will fill later: a per-segment position vector
(cumulative-ack StreamConsumer ids), the parent topic:// identity, and
cross-topic vectors (namespace consumers). The producer path fills only
v4MessageId + segmentId. MessageIdFactory (the friend the public header
already declared) is how lib/st mints ids and reaches their state.
- toByteArray()/fromByteArray() implement the MessageIdV5 wire layout
(big-endian: segmentId, classic-id bytes, position vector, parent topic
with -1-absent sentinel, multi-topic vector; trailing sections optional on
read). Malformed input returns an empty (falsy) id on the non-throwing
surface instead of throwing.
- Ordering compares (segmentId, then classic id), matching MessageIdV5;
empty ids sort first and equal each other. operator<< renders
{segment=..., id=..., positions=...}. std::hash combines the segment and
classic-id fields, completing the P6 unordered-container support.
- earliest()/latest() wrap the classic sentinels with the no-segment marker.
tests/st/StMessageIdTest.cc: sentinels, within/across-segment ordering,
empty-id semantics, unordered_set usage, stream output, producer-path and
full-section byte roundtrips, malformed/truncated input, empty
serialization. 83/83 pulsar-st-tests green; gcc-13 parity; clang-format-11
clean.
Signed-off-by: Matteo Merli <mmerli@apache.org>
Scalable-topics per-segment producers should connect straight to the DAG-provided owner broker instead of doing a fresh topic lookup. The connection machinery already supports this (HandlerBase::grabCnx(url) -> getConnection -> ClientImpl::connect), so this only threads an optional assignedBrokerUrl from createSegmentProducerAsync down to the initial connect: - HandlerBase::start(assignedBrokerUrl) overload; start() delegates to it with nullopt (grabCnx() is already grabCnx(std::nullopt), so existing handlers are unaffected). - ProducerImpl stores assignedBrokerUrl_ and passes it to the initial HandlerBase::start(); a defaulted trailing ctor param keeps every other caller unchanged. - createSegmentProducerAsync / createProducerAsyncImpl / handleCreateProducer thread the optional through (default nullopt; the public producer path is unchanged).
Define the impl side of the already-declared detail::ProducerCore so a real producer can be wired behind it: - lib/st/ProducerImplBase.h: the abstract pulsar::st::ProducerImplBase the core forwards to (six methods matching ProducerCore; matches the existing forward-decl in detail/ProducerCore.h). - lib/st/ProducerCore.cc: out-of-line definitions of the six ProducerCore methods, each a one-line forward to the impl. Required now that a producer will actually construct a core (a PULSAR_PUBLIC class must not leave exported methods undefined — cf. the Transaction/Windows-DLL fix). - ProducerCore.h: friend pulsar::st::ClientImpl so createProducerAsync can mint a core from the impl pointer.
Implement the scalable-topics producer behind detail::ProducerCore, a port of the Java v5 ScalableTopicProducer: - Owns one DagWatchSession per topic and one SegmentRouter. Per active segment, a classic pulsar::Producer is created via createSegmentProducerAsync, pinned to the DAG-provided owner broker, cached as its creation Future so concurrent cold senders share one attempt. - sendAsync routes each OutgoingMessage by key (round-robin when keyless) to a segment, publishes via the classic producer, and mints the returned id with the segment id into a pulsar::st::MessageId. Route failure is terminal; a sealed/terminated segment retries with backoff (<=10, min(100*(n+1),500)ms) and re-routes onto the layout the DAG watch delivers. - Layout changes retire producers for departed segments; exclusive access modes eagerly attach (and fail create() up front if the claim is refused), Shared attaches lazily on first send. - lastSequenceId maxes ready producers over the configured initial; flush awaits the in-flight send snapshot; close is idempotent (dag watch + all producers). Correctness details baked in from the design review: per-segment config is built FRESH each time (ProducerConfiguration's copy ctor shares its impl); the current layout is a never-null default (route yields ResultServiceUnitNotReady, not a null deref); access modes are mapped by NAME (the st and classic enums differ in value); a transactional send fails with ResultOperationNotSupported (never asserts); the in-flight entry is inserted before dispatch (warm sends complete synchronously); the layout listener captures a weak_ptr (no ownership cycle). No per-segment dispatch chain is needed — st::Future fires listeners in FIFO registration order, so same-segment sends dispatch in call order for free; a FutureTest locks that guarantee in. StClientImpl::createProducerAsync now builds a real producer (the not-supported stub and its test are gone).
Cover the pure, connection-independent logic through a small friend accessor: - Per-segment config isolation — the regression guard for the shallow ProducerConfiguration copy: building a second segment's config must not rename or re-mark the first (fresh config per segment). - Access-mode mapping by name — st and classic ProducerAccessMode enums differ in numeric value, so WaitForExclusive / ExclusiveWithFencing must not be cast. - Segment-gone classification — TopicTerminated / AlreadyClosed retry; ServiceUnitNotReady and ConnectError do not (no retry-loop on a transient routing miss). - Minted MessageId carries the real segment id.
dispatchSend/handleSegmentFailure took the user's Promise by value but only capture-copy it into their continuation lambdas — const-ref silences performance-unnecessary-value-param (the lambdas still capture their own copy).
Manual edits to lib/ClientImpl.cc (broker-pin threading) and the ProducerCore.cc include order were not clang-format-11-clean (CI's Formatting Check uses 11, local is 18). Reformat just those touched regions.
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
1 participant
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
Second implementation PR for the scalable-topics C++ SDK (
pulsar::st), building on the protocol foundation merged in #600. It adds the producer: publishing to a scalable topic by fanning out to per-segment producers, driven by the DAG-watch layout and routing keys to segments exactly like the Java v5 client (ScalableTopicProducer), so the two clients place keys on the same segments.Per PIP-468 a scalable topic's segments are ordinary persistent topics, so the produce data path reuses the classic
ProducerImplwholesale — one per active segment — andlib/stadds only the fan-out, routing, and id-mapping on top.Modifications
One commit per step, in dependency order:
segment://topic domain (classic client) —TopicNamelearns the segment domain (always v2;<parent>/<hhhh-hhhh-id>; persistent;isSegment()), and all public entry points reject segment topics;ClientImpl::createSegmentProducerAsync()is thelib/stbypass that may target one.st
MessageId— segment-qualified ids:MessageIdImpl(v4 id + segment id + position vectors) and theMessageIdV5big-endian wire layout, field-for-field compatible with the Java client for id interop. The producer fills only the v4 id + segment id.Broker-pin seam — thread an optional
assignedBrokerUrlfromcreateSegmentProducerAsyncthroughProducerImplto its initialgrabCnx, so a per-segment producer connects straight to the DAG-provided owner broker instead of doing a fresh lookup.HandlerBase::start()delegates to a newstart(assignedBrokerUrl)overload (grabCnx()already equalsgrabCnx(nullopt), so other handlers are unaffected); every new parameter defaults tonullopt, leaving the public producer path unchanged.Producer seam —
pulsar::st::ProducerImplBase(the abstract impl the publicdetail::ProducerCoreforwards to) +ProducerCore.cc(the out-of-line forwarders, required once a core is actually minted from aPULSAR_PUBLICclass).StProducerImpl— the producer, a port of Java v5ScalableTopicProducer. It owns oneDagWatchSession+ oneSegmentRouterper topic; per active segment it creates a classicpulsar::Producer(pinned to the segment's broker) cached as its creationFutureso concurrent cold senders share one attempt.sendAsyncroutes each message by key (round-robin when keyless) to a segment, publishes it, and wraps the returned classic id with the segment id into apulsar::st::MessageId. Route failure is terminal; a sealed/terminated segment retries with backoff (≤10,min(100·(n+1), 500)ms) and re-routes onto the layout the DAG watch delivers. Layout changes retire producers for departed segments; exclusive access modes eagerly attach (and failcreate()up front if the claim is refused), Shared attaches lazily.lastSequenceId/flush/closebehave as documented.No per-segment dispatch chain is needed —
st::Futurefires its listeners in FIFO registration order, so same-segment sends dispatch in call order for free (Java needs the chain only becauseCompletableFuturegives no such guarantee); aFutureTestlocks that in.Broker-free unit tests — per-segment config isolation (a regression guard:
pulsar::ProducerConfiguration's copy constructor shares its impl, so each segment's config is built fresh, not copy-then-mutated), access-mode mapping by name (the st and classic enums differ in numeric order), segment-gone classification, and MessageId minting.What works after this PR
client.newProducer<T>().topic("segment://…").create()returns a live producer;send/sendAsyncroute by key to the owning segment, publish through a per-segment classic producer pinned to the DAG broker, and return a segment-qualifiedMessageId;lastSequenceId,flush, andclosework; layout changes retire departed segments and (for exclusive modes) eagerly attach; a send whose segment was sealed transparently retries and re-routes.Next phases (follow-up PRs)
End-to-end validation against a scalable-topics broker (produce keyed + keyless, verify per-segment routing, segment ids, and message counts), then the Queue/Stream/Checkpoint consumers, the C API, and transactions. Entry-bucket batching (PIP-486) is deferred until a batcher-builder seam exists — today only the encryption-disables-batching branch is ported.
Verifying this change
pulsar-st-tests:Expected/Future(incl. the FIFO-listener ordering the producer relies on), the segment layout/router, and the producer's config isolation, access-mode mapping, segment-gone classification, and id minting.lib/stTUs and the changed classic files.Documentation
docdoc-not-needed