merlimat opened a new pull request, #601:
URL: https://github.com/apache/pulsar-client-cpp/pull/601
### 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 `ProducerImpl` wholesale — one per
active segment — and `lib/st` adds only the fan-out, routing, and id-mapping on
top.
### Modifications
One commit per step, in dependency order:
1. **`segment://` topic domain** (classic client) — `TopicName` learns the
segment domain (always v2; `<parent>/<hhhh-hhhh-id>`; persistent;
`isSegment()`), and all public entry points reject segment topics;
`ClientImpl::createSegmentProducerAsync()` is the `lib/st` bypass that may
target one.
2. **st `MessageId`** — segment-qualified ids: `MessageIdImpl` (v4 id +
segment id + position vectors) and the `MessageIdV5` big-endian wire layout,
field-for-field compatible with the Java client for id interop. The producer
fills only the v4 id + segment id.
3. **Broker-pin seam** — thread an optional `assignedBrokerUrl` from
`createSegmentProducerAsync` through `ProducerImpl` to its initial `grabCnx`,
so a per-segment producer connects straight to the DAG-provided owner broker
instead of doing a fresh lookup. `HandlerBase::start()` delegates to a new
`start(assignedBrokerUrl)` overload (`grabCnx()` already equals
`grabCnx(nullopt)`, so other handlers are unaffected); every new parameter
defaults to `nullopt`, leaving the public producer path unchanged.
4. **Producer seam** — `pulsar::st::ProducerImplBase` (the abstract impl the
public `detail::ProducerCore` forwards to) + `ProducerCore.cc` (the out-of-line
forwarders, required once a core is actually minted from a `PULSAR_PUBLIC`
class).
5. **`StProducerImpl`** — the producer, a port of Java v5
`ScalableTopicProducer`. It owns one `DagWatchSession` + one `SegmentRouter`
per topic; per active segment it creates a classic `pulsar::Producer` (pinned
to the segment's broker) cached as its creation `Future` so concurrent cold
senders share one attempt. `sendAsync` routes each message by key (round-robin
when keyless) to a segment, publishes it, and wraps the returned classic 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.
`lastSequenceId`/`flush`/`close` behave as documented.
No per-segment dispatch chain is needed — `st::Future` fires its
listeners in FIFO registration order, so same-segment sends dispatch in call
order for free (Java needs the chain only because `CompletableFuture` gives no
such guarantee); a `FutureTest` locks that in.
6. **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`/`sendAsync` route by key to the owning segment, publish
through a per-segment classic producer pinned to the DAG broker, and return a
segment-qualified `MessageId`; `lastSequenceId`, `flush`, and `close` work;
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
- **88 broker-free `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.
- Full local build (vcpkg, macOS); **clang-format-11 clean**; **clang-tidy
(LLVM 18 / libstdc++)** clean on the `lib/st` TUs and the changed classic files.
- The full produce/route/retry behavior gets its end-to-end validation
against a 5.0.0-M1 broker in the follow-up.
### Documentation
- [ ] `doc` <!-- Your PR contains doc changes -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
--
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]
