Scalable Topics: pulsar::st queue consumer — per-segment fan-in over a mux receive queue - #605
Conversation
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.
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).
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.
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.
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.
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.
Mirror the producer e2e (apache#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.
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Adds the scalable-topics (pulsar::st) queue consumer implementation with per-segment fan-in over a bounded mux receive queue, plus classic-client support for terminated topics and new tests.
Changes:
- Implemented
QueueConsumerImplwith per-segment classic consumers and a boundedReceiveQueuefor fan-in + back-pressure. - Added classic handling for
CommandReachedEndOfTopicand surfacedResultTopicTerminatedon async receive after backlog drain. - Added E2E tests for ST queue consumer and a unit test for terminated-topic async receive behavior.
Reviewed changes
Copilot reviewed 17 out of 17 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/st/StQueueConsumerE2ETest.cc | New broker-backed E2E tests for queue consumer round-trip and split-segment fan-in |
| tests/ConsumerTest.cc | Adds regression test ensuring async receive returns ResultTopicTerminated after terminate + drain |
| lib/st/StClientImpl.cc | Wires subscribeQueueAsync to the new QueueConsumerImpl |
| lib/st/ReceiveQueue.h | Declares bounded mux receive queue API used for per-segment fan-in |
| lib/st/ReceiveQueue.cc | Implements bounded offer/receive semantics with timeout support |
| lib/st/QueueConsumerImpl.h | Declares scalable-topics queue consumer core implementation |
| lib/st/QueueConsumerImpl.cc | Implements per-segment subscribe, receive loops, ack routing, and layout reconciliation |
| lib/st/QueueConsumerCore.cc | Adapts internal QueueConsumerImpl to public QueueConsumerCore API |
| lib/st/MessageImpl.h | Introduces internal received-message wrapper (classic Message + segment-qualified MessageId) |
| lib/st/MessageCore.cc | Forwards MessageCore accessors to MessageImpl |
| lib/ConsumerImpl.h | Adds reachedEndOfTopic() hook and termination flag |
| lib/ConsumerImpl.cc | Implements termination handling and surfaces ResultTopicTerminated on async receive |
| lib/ClientImpl.h | Adds subscribeSegmentAsync and an internal allow-segment flag for subscription |
| lib/ClientImpl.cc | Implements subscribeSegmentAsync and guards segment-topic rejection behind a flag |
| lib/ClientConnection.h | Declares handler for CommandReachedEndOfTopic |
| lib/ClientConnection.cc | Dispatches REACHED_END_OF_TOPIC and routes it to the target consumer |
| include/pulsar/st/detail/QueueConsumerCore.h | Allows st::ClientImpl to mint QueueConsumerCore |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…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.
lhotari
left a comment
lhotari
left a comment
There was a problem hiding this comment.
The fan-in design here is the right shape: offer() gating the segment re-arm gives real back-pressure rather than an unbounded buffer, and subscribing sealed segments as well as active ones is exactly the subtlety a split introduces. The subscribeSegmentAsync seam is properly confined too — allowSegmentTopic defaults to false, and only the st queue consumer passes true, so the public subscribe path is genuinely unchanged.
My concerns cluster on the sealed-segment drain path, which is both the most intricate new logic and the least exercised by the tests.
The drain path closes a segment's consumer as soon as the broker reports end-of-topic, but end-of-topic only means the classic prefetch queue drained — up to 1000 messages from that segment may still be sitting in the mux queue or in the application's hands. Their acks are then silently dropped, and drainedSegments_ guarantees the segment is never reattached. This is the half of the earlier Copilot thread that the drainedSegments_ fix did not cover: it cures the duplicate redelivery, but the acks are still lost.
Separately, the segment receive loop recurses rather than iterates. ConsumerImpl::receiveAsync calls back inline when a message is already buffered, offer() returns an already-completed future whenever the queue has room, and SharedState::addListener runs listeners inline — so each message adds stack frames instead of unwinding. An actively-receiving application makes this worse, not better, because offer() then hands messages straight to the parked receiver and the buffer never fills to apply the brake.
On testing: neither e2e test ever drains a sealed segment. testConsumeAcrossSplitSegments splits before producing, so the sealed parent is empty — the scenario the PR description leads with ("a split seals the parent without migrating its backlog") has no coverage. That is where I would put the next test, and it would likely surface the first finding directly.
Two gaps in the CommandReachedEndOfTopic handling, from apache#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.
… 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.
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.
…s 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 apache#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 apache#602 pinned the docker ones. Pin it to the commit the v4.11.0 tag points to (f62da5e, unchanged behavior).
lhotari
left a comment
lhotari
left a comment
There was a problem hiding this comment.
Thanks — all nine points are addressed, and several of the fixes are better than what I asked for.
The drain path (QueueConsumerImpl.cc:262-290, 308-333) is now correct. The outstanding_ refcount is incremented in onMessageFannedIn before the message is offered to the mux queue and decremented only when the application settles it, so an end-of-topic arriving with messages still in flight parks the segment in terminatedSegments_ and keeps its consumer in segmentConsumers_ for ack routing; the close runs from onMessageSettled once the count reaches zero. Settling after the ack is enqueued rather than before is the right ordering, and takeDrainedSegmentLocked handing the future back to be closed outside the lock is a nice touch.
The receive loop (QueueConsumerImpl.cc:294-297) now hops through executor_->postWork, so it is a loop rather than recursion. That was the crash-class one.
The terminated-topic flag is cleared on each new broker session (ConsumerImpl.cc:344), right alongside incomingMessages_.clear() — exactly where it needed to go, and the comment explains why termination does not cancel redelivery.
Sync receive now checks the flag in both receiveHelper overloads plus after a timeout wake-up (ConsumerImpl.cc:1254, 1288, 1301), and ConsumerTest asserts both sync paths agree with the async one.
Also fixed: the bounded subscribe retry with backoff and a still-wanted check (subscribeSegmentWithRetry), topic() now reporting the scalable topic, dead-lettering failing loudly with ResultOperationNotSupported instead of silently ignoring the policy, and the timed-receive timer now cancelled on delivery and on close.
testDrainSealedSegmentBacklog is the test I was hoping for. Producing before the split so the backlog is stranded in the sealed parent, 1200 messages to overrun both the prefetch and mux queues, asserting every message carries segment id 0, and then reattaching a second consumer to prove the acks stuck — that last assertion is what makes it a real regression test for the ack-loss bug rather than a happy-path walk.
Two non-blocking notes left in the threads rather than held against the PR: the per-call (rather than per-message) decrement in onMessageSettled, and the parked-pop() residual you documented in receiveHelper. Both are follow-up material at most.
I also checked whether closing a terminated segment on nack strands the redelivery, and I do not think it does: the receive loop has already returned by then, so an attached consumer would not deliver it either — the broker releasing it back to the subscription is the better outcome. Your inline comment gets this right.
One merge-readiness item that is not about your code: the PR validation workflow hit a startup_failure on 60c763b (run 33213225434, "This run likely failed because of a workflow file issue"), so only CodeQL and Analyze(cpp) actually ran on this head — the build and unit tests did not. It passed on the previous head, and nothing here touches workflow files, so it looks like infrastructure. Worth a re-run before merge so the classic ConsumerTest changes are actually exercised in CI.
Rebasing onto main would also pick up #604, which landed after this branch.
|
Following up on the CI note in my review — I dug into why Two Docker actions in our workflows were pinned to SHAs that are not on the ASF org-wide allowlist:
When a workflow references a non-allowlisted action, the run fails with a bare "Startup failure" — per ASF infra, "no logs, no notifications, and the PR may appear green because no checks ran." That is exactly what happened here: #610 fixes it repo-wide, bumping both to the latest approved revisions from Once #610 is merged, a rebase here should give this PR a real CI run. My approval stands — I would just want to see the build and tests actually go green on the rebased head before it lands. |
…ions (#610) Two Docker actions were pinned to SHAs that are not on the ASF org-wide allowlist: docker/setup-buildx-action@4d04d5d9 docker/setup-qemu-action@ce360397 A workflow referencing a non-allowlisted action fails with a bare "Startup failure": no logs, no notifications, and the PR can look green because no checks ran. PR #605 hit exactly this — its "PR validation" run never started, so the build and unit tests did not execute while CodeQL and Analyze(cpp) still reported success. Bump both to the latest approved revisions from apache/infrastructure-actions, and record the tag in a trailing comment so Dependabot can track them: docker/setup-buildx-action -> 37fe6310 # v4.3.0 docker/setup-qemu-action -> 96fe6ef7 # v4.2.0 docker/build-push-action 53b7df96 # v7.3.0 (already latest) Also update the GitHub-owned actions, which the ASF allowlist treats as implicitly trusted, to their latest majors: actions/checkout v3, v4 -> v7 actions/cache v3, v4 -> v6 actions/upload-artifact master -> v7 github/codeql-action v3 -> v4 jidicula/clang-format-action v4.11.0 -> v4.18.0 (allowlisted via *) actions/upload-artifact was tracking @master, an unpinned moving target. All four call sites already use unique artifact names, so the v4+ one-artifact-per-name rule is satisfied.
…-consumer # Conflicts: # .github/workflows/ci-pr-validation.yaml
Motivation
Next step in the
pulsar::st(scalable-topics) SDK after the producer (#601) and its e2e tests (#603): the queue consumer for a single scalable topic.A scalable topic is a DAG of per-segment backing topics. A queue consumer runs one
DagWatchSessionand a Shared-subscription classicpulsar::Consumerper segment — active and sealed (a split seals the parent without migrating its backlog, so its messages are only drainable through the sealed segment). Each segment's receive loop stamps the segment id onto every message and fans it into a shared, bounded mux receive queue; the user receives from that queue. Acks route back to the owning segment by the message id's segment id. Layout changes add consumers for new segments and close ones that leave the DAG.Scope here is the single-topic core (mirrors the Java v5
ScalableQueueConsumer). Dead-letter and namespace-subscription modes are deferred to follow-ups.What's in this PR
lib/st/MessageImpl+MessageCoreaccessors (wrap a classicMessage+ the segment-qualifiedMessageId).ClientImpl::subscribeSegmentAsync(a single-topic subscribe that accepts asegment://topic), guarded so the public path is unchanged.lib/st/ReceiveQueue(bounded fan-in mux with back-pressure) andlib/st/QueueConsumerImpl(per-segment consumers, ack/nack routing by segment id, layout reconcile, sealed-segment drain-and-close), wired throughQueueConsumerCoreandStClientImpl::subscribeQueueAsync.ClientConnection+ConsumerImpl.tests/st/StQueueConsumerE2ETest.cc: a produce→consume round-trip and a cross-segment fan-in test over a split topic, each self-contained (creates and splits its own topic via the admin REST API, same as the producer e2e).Note: a necessary classic-client change
The classic C++ client never handled
CommandReachedEndOfTopic(BaseCommandtype 27) — it fell through todefault:and closed the connection as an "invalid message from server". A queue consumer that subscribes to sealed segments would therefore churn its connection (close → reconnect → re-subscribe → end-of-topic again) instead of learning the segment drained.This PR handles it: dispatch the command to the target consumer (mirroring
handleActiveConsumerChange), and haveConsumerImplsurfaceResultTopicTerminatedon the async receive path once the prefetch queue drains — matching the Java client, whose consumers close a drained sealed segment onTopicTerminated. The broker only sends the command once the consumer's read position reaches the terminate marker, so buffered messages always drain first. The change is additive (a flag that is only ever set on a terminated topic), and it's covered by a newConsumerTest.testReceiveAsyncAfterTopicTerminated.Testing
pulsar-st-tests: 95/95 green (88 broker-free unit + 5 producer e2e + 2 consumer e2e) against a5.0.0-M1standalone.ConsumerTest34/34 andProducerTestterminated tests 2/2 green (no regression from the terminated-topic change).Follow-ups
StProducerE2ETest.cc(landed in Scalable Topics: producer end-to-end tests + runpulsar::sttests in CI #603) andStQueueConsumerE2ETest.cc; a small follow-up can extract a sharedtests/st/StE2EAdmin.h.