Scalable Topics: pulsar::st stream consumer — controller-assigned ordered consumption with cumulative acks

[merlimat]

Motivation

Next consumer in the pulsar::st (scalable-topics) SDK after the queue consumer (#605): the stream consumer — ordered consumption over the segment DAG.

The design headline, from the Java v5 reference (ScalableStreamConsumer / ScalableConsumerClient / broker SubscriptionCoordinator): the client enforces no DAG ordering itself. The broker's subscription coordinator withholds an active child segment from the consumer's assignment until every parent has drained for this subscription (per-subscription backlog reaching zero; merges fall out of the same rule). The client's job is therefore to register with the controller, subscribe Exclusive to exactly the assigned segments, and mux them — per-key order across splits and merges holds as long as the application keeps acknowledging. That makes acking-as-you-go a liveness requirement: a consumer that defers all acks to the end keeps the children withheld forever (documented on the API and encoded in the e2e).

What's in this PR

• Consumer assignment session — the controller-registration protocol the queue consumer never needed: Commands::newScalableTopicSubscribe, ClientConnection registries for pushed CommandScalableTopicAssignmentUpdate (by consumer id) and one-shot CommandScalableTopicSubscribeResponse callbacks (by request id), both drained on connection close — the same idiom as the DAG-watch session. ConsumerAssignmentSession resolves the controller leader through a one-shot DAG-watch lookup (TLS-aware; behind a proxy or before leader election it falls back to the regular lookup path, where any broker forwards), registers + subscribes, applies epoch-gated assignments, replays the current assignment on listener registration, reconnects with backoff after the initial assignment and fails fast before it. close() drops only the local registration; the broker reaps it through its grace timer (Java parity).
• StreamConsumerImpl — one Exclusive classic consumer per assigned segment, receive loops fanning into the shared mux ReceiveQueue, and every delivered message stamped with a position vector: a snapshot of all segments' latest-delivered ids taken at delivery time, so one cumulative acknowledgment advances every cursor (fanned out as one acknowledgeCumulativeAsync per segment). TopicTerminated closes the segment immediately and late acks no-op — a deliberate divergence from the queue consumer's deferred close, which doesn't transfer because one cumulative ack settles an unbounded prefix. Newly assigned segments retry only rebalance collisions (ConsumerBusy / ConsumerAssignError, the previous Exclusive owner not having released yet) with bounded backoff, and fail fast on anything else. AckPolicy::negativeAckRedeliveryDelay is deliberately not wired — a stream consumer has no negative-ack path.
• receiveMulti — ReceiveQueue::receiveMultiAsync (greedy drain, then wait with the remaining deadline until full or timeout; a short or empty batch is a normal result), with broker-free unit tests. This also closes G1 from the API review for the stream side.
• Seam fix — the readCompacted guard only accepted the persistent:// domain and would have rejected every segment:// subscribe; relaxed only on the internal allowSegmentTopic path.
• e2e (runs under the existing st CI phase with no workflow changes) — three tests against a real broker:
  • ordered round-trip on one segment + one cumulative ack of the last message settles the stream (a reattached consumer receives nothing);
  • the DAG-replay scenario: produce → split → produce → attach Earliest; the initial assignment carries only the sealed parent, acking-as-we-go drains it, the broker's drain detection pushes the children, and every parent message arrives before any child message;
  • two initial segments drained via receiveMulti with no intermediate acks; acknowledging only the final message advances both cursors through its position vector.
• Harness dedup — the admin REST helpers the three e2e suites each carried now live in tests/st/StE2EAdmin.h (the follow-up flagged in Scalable Topics: pulsar::st queue consumer — per-segment fan-in over a mux receive queue #605).

Deferred (fails loudly, not silently)

• PIP-486 bucket-shared assignments (consumers outnumbering segments): the subscribe fails with ResultOperationNotSupported when the initial assignment carries bucket ranges. The whole-segment Exclusive path covers every assignment the controller produces otherwise, and the deferral avoids porting the drain-release machinery blind.
• Namespace-mode subscriptions, same as the queue consumer.

Testing

• pulsar-st-tests: 105/105 — 94 broker-free unit tests (incl. new session + receiveMulti suites) and 11 e2e (5 producer + 3 queue + 3 stream) against a 5.0.0-M1 standalone.
• clang-format-11 and clang-tidy (the Lint job's checks) clean on all new/changed translation units.