Scalable Topics: producer end-to-end tests + run pulsar::st tests in CI - #603
Merged
BewareMyPower merged 6 commits intoJul 20, 2026
Merged
Conversation
Validate the producer against a real scalable-topics broker and run the st tests in CI (they were built but never executed before). - tests/st/StProducerE2ETest.cc: end-to-end producer tests against a standalone broker — keyed + keyless publishing that fans out to segments, asserting each send returns a segment-qualified MessageId, lastSequenceId advances, and flush/close succeed; plus an async-batch case exercising the in-flight tracking. Gated on PULSAR_ST_E2E so the ordinary broker-free run skips them. - tests/st/docker-compose.yml: a plain apachepulsar/pulsar:latest (5.0.0-M1) standalone broker; it ships the scalable-topics controller and wire protocol with no extra config. - run-unit-tests.sh: bring the broker up, create the scalable topic the producer publishes to (scalable topics are a managed construct — not auto-created on lookup like regular topics), then run the full pulsar-st-tests (88 broker-free cases + the e2e), and tear the broker down. Verified locally: the whole produce path works against the M1 broker (DAG-watch lookup, key routing, per-segment producer creation, MessageId minting); 89 tests pass with the broker up, and the e2e cases skip cleanly without it.
The scalable topic is pre-created not because scalable topics are inherently managed/non-auto-created, but because the pinned 5.0.0-M1 broker image has a bug where create_if_missing does not auto-create on lookup — already fixed in master. Reword the comment so the pre-create can be dropped once the image carries the fix.
The existing e2e cases produce to a single-segment topic, so every key routes to that one segment and cross-segment routing is never exercised. Add a case that publishes to a topic the harness has split into two active half-range segments and asserts the produced message ids span both segments. The harness creates the topic and splits segment 0 (which seals it and yields two active children) before the run.
Add two more producer e2e cases so split/merge topology changes are exercised against a real broker: - testProduceAfterMerge: publish to a topic the harness split then merged back. A merge seals both children and creates one full-range active segment, so every key routes to it — this drives the layout parser and router on a DAG that carries several sealed segments. - testProduceContinuesAcrossLiveSplit: warm up a per-segment producer, then seal that segment (split it) out from under the producer while a burst of async sends is in flight. Sends that hit the sealed segment fail with TopicTerminated and the producer retries + re-routes onto the layout the DAG watch delivers; every send must ultimately succeed and publishing must reach the post-split segments. The split is triggered through the admin command the harness passes in PULSAR_ST_E2E_SPLIT_CMD, keeping the container name out of the test. The harness creates the merge topic (create -> split -> merge) and the live-split topic, and exports the split command.
CI pulled apachepulsar/pulsar:latest, which is a stable release whose pulsar-admin
does not know the scalable-topics command ("Unmatched arguments"), so topic setup
failed. The moving :latest tag does not carry scalable topics yet; the published
5.0.0-M1 milestone does. Pin to it. (A local :latest tagged as an M1 snapshot is
what made this pass locally — that tag is not what CI pulls.)
Address review feedback: the producer end-to-end tests arranged their segment
topology out-of-band — run-unit-tests.sh pre-split/merged topics with
pulsar-admin, and the live-split test shelled out via std::system on a
PULSAR_ST_E2E_SPLIT_CMD command passed in from the harness. Move all of it into
the tests, which now create their own topic and trigger split/merge at the exact
point they want through the scalable-topics admin REST API (PUT .../scalable/
{tenant}/{ns}/{topic}, POST .../split/{seg}, POST .../merge/{id1}/{id2}).
Each test uses a fresh topic name so runs are self-contained and cannot wedge a
reused broker. run-unit-tests.sh drops all scalable-topics CLI setup and the
PULSAR_ST_E2E_SPLIT_CMD env var; it just waits for the broker to accept admin
requests. HttpHelper.cc (the tests' REST client) is linked into pulsar-st-tests.
All five e2e cases pass against apachepulsar/pulsar:5.0.0-M1.
BewareMyPower
approved these changes
Jul 20, 2026
merlimat
added a commit
that referenced
this pull request
Aug 29, 2026
…r a mux receive queue (#605) * st: received-message plumbing — MessageImpl + MessageCore accessors The consumer receive path needs the impl side of detail::MessageCore, which was declared but never defined (like ProducerCore was before the producer landed): - lib/st/MessageImpl.h: pulsar::st::MessageImpl, a thin view over a classic pulsar::Message (owns payload + metadata) plus the segment-qualified st MessageId minted on receive, with an optional topic override for namespace mode. - lib/st/MessageCore.cc: the out-of-line MessageCore accessors, forwarding to it. All accessors map to public classic Message getters except sequenceId(), which the classic public API does not expose; it returns -1 for now (a TODO to revisit with a classic accessor when the Stream consumer needs it, rather than touch the classic API here). Shared by all three consumer types. * st: classic consumer segment seam — subscribeSegmentAsync The scalable-topics queue/stream consumers attach a Shared consumer per active segment, on the segment's segment:// backing topic — which the public subscribe path rejects. Add ClientImpl::subscribeSegmentAsync, mirroring the producer's createSegmentProducerAsync: the private single-topic subscribeToTopicsAsyncV2 gains an allowSegmentTopic flag (default false; the segment-domain rejection becomes isSegment() && !allowSegmentTopic), and the new public method calls it with true. A segment is a non-partitioned persistent topic, so it lands in the single-ConsumerImpl branch of handleSubscribe unchanged. No broker pin (the Java consumer path does not pin; the DAG-provided owner resolves via segment:// lookup). * st: clang-format-11 line wrapping in segment seam + MessageImpl ctor Wrap two over-length lines that clang-format-11 (the CI style) breaks but clang-format-18 leaves on one line: the subscribeToTopicsAsyncV2 call in ClientImpl::subscribeSegmentAsync and the MessageImpl constructor signature. Formatting only, no behavior change. * st: queue consumer core — per-segment fan-in over a mux receive queue Implement the single-topic scalable-topics queue consumer (a port of the Java v5 ScalableQueueConsumer). A Shared subscription is fanned across every segment of the topic — active AND sealed, since a sealed segment may still hold undrained messages — with one classic Shared-subscription pulsar::Consumer per segment created through the ClientImpl::subscribeSegmentAsync seam. - ReceiveQueue: a bounded fan-in mux. Per-segment receive loops offer() messages; the user receiveAsync()es them in FIFO order. offer() returns a future that completes only when the queue has room, so a slow consumer back-pressures the underlying segment consumers' flow control rather than buffering unboundedly. Timed receives fail with ResultTimeout; close() fails every waiter. - QueueConsumerImpl: owns one DagWatchSession and the per-segment consumers. Each segment loop stamps the segment id onto every message (MessageIdFactory) and fans it into the shared queue. Acks/nacks route back to the owning segment's consumer via the message id's segment id. Layout changes add consumers for new segments and close ones that left the DAG; a segment that reports TopicTerminated (a drained sealed segment) is closed and dropped. - QueueConsumerCore: thin forwarders mapping MessageImplPtr to MessageCore. - Wire ClientImpl::subscribeQueueAsync (was notImplementedYet), mirroring createProducerAsync: build the impl, start(), then mint the public core. Transactional acknowledge is not implemented yet (logged and dropped); the dead-letter and namespace-subscription paths are deferred to later slices. * Handle CommandReachedEndOfTopic on the consumer receive path The classic client never handled BaseCommand::REACHED_END_OF_TOPIC (type 27): handleIncomingCommand fell through to default: and closed the whole connection as an "invalid message from server". Any consumer of a terminated topic — and every scalable-topics queue consumer, which subscribes to sealed segments to drain their backlog — would therefore churn its connection (close, reconnect, re-subscribe, reach end of topic again) instead of learning the topic ended. Handle it: dispatch REACHED_END_OF_TOPIC to the target consumer (mirroring handleActiveConsumerChange), and have ConsumerImpl surface ResultTopicTerminated on the async receive path once the prefetch queue drains — matching the Java client, whose consumers close a drained sealed segment on TopicTerminated. The broker only sends the command once the consumer's read position reaches the terminate marker, so buffered messages always drain before termination. Scope is the async receiveAsync path (what the scalable consumer uses); the blocking sync receive() is unchanged (it would need to interrupt a parked pop(), and a terminated topic there already behaves as "no more messages"). Adds ConsumerTest.testReceiveAsyncAfterTopicTerminated. * st: queue consumer produce->consume e2e test End-to-end coverage for the scalable-topics queue consumer against a real broker, gated on PULSAR_ST_E2E (the broker-free unit run skips it): - testProduceThenConsumeRoundTrip: produce 25 keyed messages, receive and ack all of them through a Shared subscription, assert the payloads round-trip and every received id carries a real segment id. - testConsumeAcrossSplitSegments: over a topic pre-split into two active segments, produce 60 keyed messages and assert they fan in from both segments through the mux receive queue — the multi-segment path the queue consumer exists for, and the case that exercises draining the sealed parent segment. Both pass against apachepulsar/pulsar:5.0.0-M1. The CI wiring (docker-compose + run-unit-tests.sh) that runs these lands with the producer-e2e harness. * st: drive queue-consumer e2e split through the admin REST API Mirror the producer e2e (#603): each queue-consumer e2e test now creates its own fresh-named scalable topic — and, for the fan-in test, splits it — through the admin REST API, instead of consuming harness-pre-created, CLI-pre-split fixed topics (st-e2e-queue / st-e2e-queue-split). The tests are now self-contained and keep working under the REST-driven harness, where nothing is pre-arranged for them. Links HttpHelper.cc into pulsar-st-tests for the makePut/makePostRequest calls. * st: address #605 review — clang-tidy move + drained-segment re-subscribe - The queue-consumer subscribe callback applied std::move to a pulsar::Consumer, whose virtual destructor suppresses the move constructor, so the move bound to the copy constructor: clang-tidy performance-move-const-arg, which failed the Lint job. Copy the handle directly (a shared-impl copy), matching StProducerImpl. - On ResultTopicTerminated the drained segment's consumer was erased, but the sealed segment stays in the DAG, so the next layout reconcile re-subscribed it and the broker redelivered its still-unacked messages as duplicates. Track drained segments and skip re-subscribing them; prune the set when a segment leaves the DAG. * Terminated-topic completeness on the consumer: reconnect + sync receive Two gaps in the CommandReachedEndOfTopic handling, from #605 review: - hasReachedEndOfTopic_ was never cleared, so after a reconnect (which clears the prefetch queue and re-sends flow permits) a receiveAsync landing before redelivery arrived would report a stale ResultTopicTerminated — and the scalable queue consumer would then drop the segment permanently. Termination stops new publications, not redelivery of unacked messages: clear the flag on each new broker session; the broker re-sends the command once the re-created consumer's read position reaches the terminate marker again. - The sync receive paths ignored the flag entirely: the untimed receive() blocked forever on a drained terminated topic and the timed one returned ResultTimeout. Both now fail fast with ResultTopicTerminated when the flag is set and the queue is empty, agreeing with the async path. (A receive already parked in pop() when the command arrives still waits — waking it would need an interruptible queue.) Extends ConsumerTest.testReceiveAsyncAfterTopicTerminated to assert both sync overloads. * st: queue consumer drain, robustness, and honesty fixes from #605 review - Ack loss on drained sealed segments: end-of-topic only means the classic prefetch queue drained — messages already fanned into the mux queue or held by the application still need the segment consumer to route their acks. Track outstanding (fanned-in minus acked/nacked) messages per segment and defer the drain-close until the count reaches zero; the consumer stays in the map for ack routing meanwhile. - Receive-loop recursion: receiveAsync completes inline when a message is prefetched and offer()'s future is already complete while the queue has room, so the re-arm chain grew the stack once per message. Hop the re-arm through the IO executor so the chain is a loop again. - A segment subscribe that failed off the first-layout path was only retried on the next DAG push, which may not come for hours: back it with a bounded backoff retry (10 attempts, 100->500ms, the producer's constants), skipping segments that left the DAG or drained. - Message::topic() reported the internal segment:// backing topic; pass the scalable topic as the override so the public contract holds. - A configured deadLetterPolicy was silently ignored; it now fails the subscribe with ResultOperationNotSupported, and the API docs say so, until dead-lettering lands. - ReceiveQueue timed receives never cancelled their timer when a message won the race, accumulating live timers proportional to receive rate x timeout; park {promise, timer} together and cancel on delivery and on close. * st: e2e for draining a sealed segment's backlog with sticking acks The scenario the queue consumer exists for — a split seals the parent WITHOUT migrating its backlog — had no coverage: both existing e2e tests split before producing, so the sealed parent was always empty and no end-of-topic was ever delivered. Produce 1200 messages (more than the classic prefetch queue and the mux capacity), split, then consume: every pre-split message must arrive through the sealed parent, and reattaching a second consumer on the same subscription must receive nothing — proving the acks routed through the drain-deferred close instead of being dropped. Fails on the pre-fix code. * Pin jidicula/clang-format-action to its commit SHA for the ASF actions policy Since mid-August every new PR-validation run on this repo fails at workflow startup (startup_failure, 0s, "workflow file issue") with no workflow change on main — the same repo-wide pattern as the docker/build-push-action break fixed by #602. The ASF GitHub Actions policy requires external actions to be pinned to a specific git hash, and jidicula/clang-format-action@v4.11.0 was the one remaining tag-pinned external action after #602 pinned the docker ones. Pin it to the commit the v4.11.0 tag points to (f62da5e, unchanged behavior).
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 #601 (the
pulsar::stproducer). That PR shipped the producer with broker-free unit tests; this one validates it end-to-end against a real scalable-topics broker and runs thepulsar::sttests in CI for the first time (they were built but never executed).Modifications
tests/st/StProducerE2ETest.cc— end-to-end producer tests against a standalone broker. A keyed + keyless case that fans out across segments, asserting every send returns a segment-qualifiedMessageId,lastSequenceIdadvances, andflush()/close()succeed; and an async-batch case that fires manysendAsync()calls across keys and awaits them all (exercising the in-flight tracking and flush). Both are gated on thePULSAR_ST_E2Eenvironment variable, so the ordinary broker-free unit-test run skips them.tests/st/docker-compose.yml— a single standaloneapachepulsar/pulsar:latestbroker for the e2e run. The image ships the scalable-topics controller and wire protocol; no extra broker config is needed to enable them.run-unit-tests.sh— a scalable-topics phase: bring the broker up, create the scalable topic the producer publishes to, run the fullpulsar-st-tests(the 88 broker-free cases plus the e2e), then tear the broker down. This is the first place anypulsar-st-testsrun in CI.What this validates
The whole producer data path works against the real broker: the DAG-watch lookup resolves the scalable topic, keys route to segments, a per-segment producer is created on the
segment://…topic, and each publish comes back with a segment-qualifiedMessageId.Notes
apachepulsar/pulsar:latestdoes not auto-create a scalable topic on lookup (a bug fixed in later releases), so the harness pre-creates it withpulsar-admin scalable-topics create. That step is harmless once the image carries the fix — the producer'screate_if_missinglookup finds the topic either way — and can be dropped then.MessageIdon each send is the landing check (the scalable-topic admin stats expose only DAG structure, not per-segment message counts).Verifying this change
Documentation
docdoc-not-needed