merlimat opened a new pull request, #600: URL: https://github.com/apache/pulsar-client-cpp/pull/600
### Motivation This is the first implementation PR for the scalable-topics C++ SDK (`pulsar::st`), building on the API definition merged in #598. It lays the **protocol foundation** the producer/consumer implementations will sit on: the wire contract, the build integration for `lib/st`, the client shell, and the DAG-watch machinery that keeps a scalable topic's segment layout current on the client. The guiding architecture (per PIP-468): a scalable topic's segments are **ordinary persistent topics** broker-side, so the classic transport stack (connection pool, executors, lookup) is reused wholesale, and `lib/st` layers the scalable-topics machinery on top. The layout/routing/session code is a faithful port of the Java v5 client (`ClientSegmentLayout` / `SegmentRouter` / `ScalableTopicHashing` / `DagWatchClient`), so the two clients behave identically on the wire and route keys identically. ### Modifications One commit per step, in dependency order: 1. **Proto sync** — `proto/PulsarApi.proto` updated from apache/pulsar master, picking up the complete scalable-topics protocol: the DAG-watch commands (`CommandScalableTopicLookup`/`Update`/`Close`), segment/DAG messages, consumer-controller commands, topic/TC watch commands, and the `supports_scalable_topics` feature flag (BaseCommand types 70–81). 2. **Build wiring** — `lib/st` compiles as `ST_OBJECT_LIB`, a separate object library with per-target `CXX_STANDARD 20` (the classic client stays C++17), merged into the same `libpulsar` shared/static artifacts. `reflectcpp` joins the vcpkg manifest, so `jsonSchema<T>()` builds against the real reflect-cpp. New `pulsar-st-tests` gtest binary (no broker needed). 3. **Client shell** — `pulsar::st::ClientImpl` wraps a classic `pulsar::ClientImpl`; `PulsarClientBuilder::build()` validates the config and maps the st policy structs onto `ClientConfiguration`; `detail::ClientCore` methods forward to it; `closeAsync()` bridges the classic callback onto the continuation `Future<void>`. Operations whose phase hasn't landed fail cleanly with `ResultOperationNotSupported` instead of hanging. 4. **Segment layout + router** — `ScalableTopicHashing` (one raw murmur3 per key; high 16 bits route segments, low 16 route entry-buckets per PIP-486 — `Murmur3_32Hash` grows a `makeRawHash()` mirroring the Java addition), `SegmentLayout::fromProto` (sorted active segments, sealed segments, per-segment broker addresses, controller address, legacy markers, entry-bucket splits, `segment://…/hhhh-hhhh-id` topic URIs), and `SegmentRouter` (range routing; all-legacy synthetic layouts fall back to `signSafeMod(classicMurmur3(key), N)` for exact v4 partitioned-producer parity; keyless messages round-robin). 5. **Wire plumbing** — `Commands::newScalableTopicLookup/Close` factories; `supports_scalable_topics` advertised on CONNECT; `messageType()` names for all twelve new command types (it runs on every incoming command and throws on unknown values); `ClientConnection` gains a DAG-watch session registry keyed by the client-assigned session id, `SCALABLE_TOPIC_UPDATE` dispatch (unknown sessions drop the push rather than killing the shared connection), close-notification for re-establishment, and now records the broker's CONNECTED feature flags (`supportsScalableTopics()`). 6. **`DagWatchSession`** — the per-topic session tying it together: connect through the classic pool, gate on broker support, register + send the lookup; updates adopt the broker-resolved `topic://` identity and swap the `SegmentLayout` (listener notified with new/old); reconnect with exponential backoff (100ms–30s) after the initial layout, fail-fast before it; broker errors before the first layout fail `start()` (`TopicNotFound` mapped); idempotent `close()` sends `CommandScalableTopicClose`. 7. **Avro backend** — the registry reflectcpp port ships `rfl/avro.hpp` but exposes no `avro` feature, so the backend wasn't compiled and `<avro.h>` wasn't installed. A vcpkg **overlay port** (`vcpkg-overlay/reflectcpp`, wired via `vcpkg-configuration.json`) adds the feature (`REFLECTCPP_AVRO` + `avro-c`); building against the real backend exposed and fixed two API mismatches in `AvroSchema.h` (`to_schema<T>().json_str()`, `write()` → `std::vector<char>`). Verified with a runtime `avroSchema<T>()` roundtrip. The overlay can be dropped once the upstream port grows the feature. ### What works after this PR `PulsarClient::builder().serviceUrl(…).build()` returns a live client on the real connection pool; `close()`/`shutdown()` work; the DAG-watch session can open, receive layouts, survive reconnects, and close. `jsonSchema` and `avroSchema` build and roundtrip against the real reflect-cpp. Producer/consumer creation intentionally returns `ResultOperationNotSupported` until their phases land. ### Next phases (follow-up PRs) Producer (per-segment fan-out driven by the watch session, keyed routing, st `MessageId`), then Queue/Stream/Checkpoint consumers, the C API, and transactions. ### Verifying this change - `pulsar-st-tests`: **70 broker-free tests** covering `Expected`/`Future` (incl. coroutine awaiter and abandoned-promise), the Schema codecs, builder validation and client lifecycle, layout parsing/routing (incl. mod-N parity with the classic hash), wire-level command roundtrips, and session lifecycle. - Full local builds of `pulsarShared`/`pulsarStatic`/`pulsar-tests`/`StExamples` (vcpkg, macOS); gcc-13 parity compiles with the repo warning flags; clang-format-11 clean. - The full session protocol and routing get their end-to-end validation against a scalable-topics broker in the producer PR. ### 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]
