Scalable Topics: pulsar::st queue consumer — per-segment fan-in over 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).

Author: merlimat

include/pulsar/st/QueueConsumer.h

@@ -81,7 +81,8 @@ struct QueueConsumerConfig {
      *  redelivery delay). Default-constructed `AckPolicy` when unset. */
     AckPolicy ackPolicy;
     /** Optional dead-letter policy: route messages to a dead-letter topic after
-     *  repeated redelivery. Default unset (no dead-lettering). */
+     *  repeated redelivery. Default unset (no dead-lettering). Not implemented yet:
+     *  setting it fails the subscribe with `ResultOperationNotSupported`. */
     std::optional<DeadLetterPolicy> deadLetterPolicy;
     /** Arbitrary client-side consumer properties (reported in topic stats). Default empty. */
     Properties properties;
@@ -328,6 +329,9 @@ class QueueConsumerBuilder {
      * Route messages to a dead-letter topic after repeated redelivery (spec §7.2).
      * QueueConsumer only.
      *
+     * Not implemented yet: setting a policy currently fails the subscribe with
+     * `ResultOperationNotSupported` rather than silently ignoring it.
+     *
      * @param policy the dead-letter policy (max redeliveries, DLQ topic name, etc.).
      *        Default unset (no dead-lettering).
      * @return `*this` for chaining.

include/pulsar/st/detail/QueueConsumerCore.h

@@ -33,6 +33,7 @@ namespace pulsar::st {
 class QueueConsumerImpl;
 using QueueConsumerImplPtr = std::shared_ptr<QueueConsumerImpl>;
 class Transaction;
+class ClientImpl;  // lib/st — mints consumer cores from subscribeQueueAsync
 
 namespace detail {
 
@@ -60,6 +61,7 @@ class PULSAR_PUBLIC QueueConsumerCore {
 
    private:
     friend class ClientCore;
+    friend class ::pulsar::st::ClientImpl;
     explicit QueueConsumerCore(QueueConsumerImplPtr impl) : impl_(std::move(impl)) {}
 
     QueueConsumerImplPtr impl_;

lib/ClientConnection.cc

@@ -853,6 +853,27 @@ void ClientConnection::handleActiveConsumerChange(const proto::CommandActiveCons
     }
 }
 
+void ClientConnection::handleReachedEndOfTopic(const proto::CommandReachedEndOfTopic& reachedEndOfTopic) {
+    LOG_DEBUG(cnxString() << "Received reached-end-of-topic, consumer_id: "
+                          << reachedEndOfTopic.consumer_id());
+    Lock lock(mutex_);
+    ConsumersMap::iterator it = consumers_.find(reachedEndOfTopic.consumer_id());
+    if (it != consumers_.end()) {
+        ConsumerImplPtr consumer = it->second.lock();
+        if (consumer) {
+            lock.unlock();
+            consumer->reachedEndOfTopic();
+        } else {
+            consumers_.erase(reachedEndOfTopic.consumer_id());
+            LOG_DEBUG(cnxString() << "Ignoring reached-end-of-topic for already destroyed consumer "
+                                  << reachedEndOfTopic.consumer_id());
+        }
+    } else {
+        LOG_DEBUG(cnxString() << "Got invalid consumer Id in reached-end-of-topic "
+                              << reachedEndOfTopic.consumer_id());
+    }
+}
+
 void ClientConnection::handleIncomingMessage(const proto::CommandMessage& msg, bool isChecksumValid,
                                              proto::BrokerEntryMetadata& brokerEntryMetadata,
                                              proto::MessageMetadata& msgMetadata, SharedBuffer& payload) {
@@ -997,6 +1018,10 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {
                     handleScalableTopicUpdate(incomingCmd.scalabletopicupdate());
                     break;
 
+                case BaseCommand::REACHED_END_OF_TOPIC:
+                    handleReachedEndOfTopic(incomingCmd.reachedendoftopic());
+                    break;
+
                 default:
                     LOG_WARN(cnxString() << "Received invalid message from server");
                     close(Error{ResultDisconnected, cnxString() + "Received invalid message from server"});

lib/ClientConnection.h

@@ -105,6 +105,7 @@ class CommandGetLastMessageIdResponse;
 class CommandLookupTopicResponse;
 class CommandPartitionedTopicMetadataResponse;
 class CommandProducerSuccess;
+class CommandReachedEndOfTopic;
 class CommandScalableTopicUpdate;
 class CommandSendReceipt;
 class CommandSendError;
@@ -265,6 +266,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
                         proto::BaseCommand& incomingCmd);
 
     void handleActiveConsumerChange(const proto::CommandActiveConsumerChange& change);
+    void handleReachedEndOfTopic(const proto::CommandReachedEndOfTopic& reachedEndOfTopic);
     void handleIncomingCommand(proto::BaseCommand& incomingCmd);
     void handleIncomingMessage(const proto::CommandMessage& msg, bool isChecksumValid,
                                proto::BrokerEntryMetadata& brokerEntryMetadata,

lib/ClientImpl.cc

@@ -604,8 +604,15 @@ void ClientImpl::subscribeAsync(const std::string& topic, const std::string& sub
                      [callback](const auto& value) { invokeLegacyCallback<Consumer>(callback, value); });
 }
 
+void ClientImpl::subscribeSegmentAsync(const std::string& topic, const std::string& subscriptionName,
+                                       const ConsumerConfiguration& conf, SubscribeV2Callback callback) {
+    subscribeToTopicsAsyncV2(topic, subscriptionName, conf, std::move(callback),
+                             /* allowSegmentTopic */ true);
+}
+
 void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::string& subscriptionName,
-                                          const ConsumerConfiguration& conf, SubscribeV2Callback callback) {
+                                          const ConsumerConfiguration& conf, SubscribeV2Callback callback,
+                                          bool allowSegmentTopic) {
     LOG_INFO("Subscribing on Topic :" << topic);
     TopicNamePtr topicName;
     {
@@ -627,7 +634,7 @@ void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::s
         }
     }
 
-    if (topicName->isSegment()) {
+    if (topicName->isSegment() && !allowSegmentTopic) {
         callback(segmentTopicRejected(topic));
         return;
     }

lib/ClientImpl.h

@@ -103,6 +103,14 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
                                     CreateProducerV2Callback callback,
                                     const std::optional<std::string>& assignedBrokerUrl = std::nullopt);
 
+    /**
+     * Subscribe a consumer to a single `segment://` scalable-topic segment, bypassing the
+     * segment-domain rejection applied to the public subscribe path. The scalable-topics
+     * queue/stream consumers use this to attach a per-segment consumer.
+     */
+    void subscribeSegmentAsync(const std::string& topic, const std::string& subscriptionName,
+                               const ConsumerConfiguration& conf, SubscribeV2Callback callback);
+
     void subscribeAsync(const std::string& topic, const std::string& subscriptionName,
                         const ConsumerConfiguration& conf, const SubscribeCallback& callback);
 
@@ -203,7 +211,8 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
                          ConsumerConfiguration conf, SubscribeV2Callback callback);
 
     void subscribeToTopicsAsyncV2(const std::string& topic, const std::string& subscriptionName,
-                                  const ConsumerConfiguration& conf, SubscribeV2Callback callback);
+                                  const ConsumerConfiguration& conf, SubscribeV2Callback callback,
+                                  bool allowSegmentTopic = false);
 
     void subscribeToTopicsAsyncV2(const std::vector<std::string>& topics, const std::string& subscriptionName,
                                   const ConsumerConfiguration& conf, SubscribeV2Callback callback);

lib/ConsumerImpl.cc

@@ -337,6 +337,11 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
             incomingMessages_.clear();
             possibleSendToDeadLetterTopicMessages_.clear();
             backoff_.reset();
+            // Re-derive end-of-topic from the new session: termination stops new publications, not
+            // redelivery of unacked messages, so a stale flag would report ResultTopicTerminated in
+            // the window before redeliveries arrive. The broker re-sends CommandReachedEndOfTopic
+            // once this consumer's read position reaches the terminate marker again.
+            hasReachedEndOfTopic_ = false;
             if (!messageListener_ && config_.getReceiverQueueSize() == 0) {
                 // Complicated logic since we don't have a isLocked() function for mutex
                 if (waitingForZeroQueueSizeMessage) {
@@ -823,6 +828,24 @@ void ConsumerImpl::activeConsumerChanged(bool isActive) {
     }
 }
 
+void ConsumerImpl::reachedEndOfTopic() {
+    hasReachedEndOfTopic_ = true;
+    // If nothing is buffered there is nothing left to deliver, so complete any waiting async
+    // receives with ResultTopicTerminated now. When messages are still buffered they drain through
+    // the normal path first, and the next receive observes the flag (see receiveAsync).
+    Lock lock(pendingReceiveMutex_);
+    if (incomingMessages_.empty()) {
+        Message msg;
+        while (!pendingReceives_.empty()) {
+            ReceiveCallback callback = pendingReceives_.front();
+            pendingReceives_.pop();
+            listenerExecutor_->postWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback,
+                                                  get_shared_this_ptr(), ResultTopicTerminated, msg,
+                                                  callback));
+        }
+    }
+}
+
 void ConsumerImpl::internalConsumerChangeListener(bool isActive) {
     try {
         if (isActive) {
@@ -1189,6 +1212,14 @@ void ConsumerImpl::receiveAsync(const ReceiveCallback& callback) {
         messageProcessed(msg);
         msg = interceptors_->beforeConsume(Consumer(shared_from_this()), msg);
         callback(ResultOk, msg);
+    } else if (hasReachedEndOfTopic_) {
+        // Terminated topic with nothing left buffered: fail the receive rather than parking it
+        // forever waiting for a message that will never arrive.
+        pendingReceiveMutexLock.unlock();
+        if (config_.getReceiverQueueSize() == 0) {
+            mutexlock.unlock();
+        }
+        callback(ResultTopicTerminated, msg);
     } else if (config_.getReceiverQueueSize() == 0) {
         pendingReceives_.push(callback);
         // If connection_ is nullptr, sendFlowPermitsToBroker does nothing.
@@ -1217,6 +1248,13 @@ Result ConsumerImpl::receiveHelper(Message& msg) {
         return fetchSingleMessageFromBroker(msg);
     }
 
+    // A drained terminated topic has nothing left to deliver: fail fast instead of blocking
+    // forever, matching the async path. (A receive already parked in pop() when end-of-topic
+    // arrives still waits — the queue only wakes on a message or on close.)
+    if (hasReachedEndOfTopic_ && incomingMessages_.empty()) {
+        return ResultTopicTerminated;
+    }
+
     if (!incomingMessages_.pop(msg)) {
         return ResultInterrupted;
     }
@@ -1247,6 +1285,10 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
         return ResultInvalidConfiguration;
     }
 
+    if (hasReachedEndOfTopic_ && incomingMessages_.empty()) {
+        return ResultTopicTerminated;
+    }
+
     if (incomingMessages_.pop(msg, std::chrono::milliseconds(timeout))) {
         messageProcessed(msg);
         msg = interceptors_->beforeConsume(Consumer(shared_from_this()), msg);
@@ -1255,6 +1297,10 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
         if (state_ != Ready) {
             return ResultAlreadyClosed;
         }
+        // Waking up empty on a terminated topic means drained, not merely idle.
+        if (hasReachedEndOfTopic_ && incomingMessages_.empty()) {
+            return ResultTopicTerminated;
+        }
         return ResultTimeout;
     }
 }

lib/ConsumerImpl.h

@@ -103,6 +103,9 @@ class ConsumerImpl : public ConsumerImplBase {
                          proto::MessageMetadata& msgMetadata, SharedBuffer& payload);
     void messageProcessed(Message& msg, bool track = true);
     void activeConsumerChanged(bool isActive);
+    // The broker signalled that this (terminated) topic has no more messages beyond what has already
+    // been delivered. Surface ResultTopicTerminated to receivers once the prefetch queue drains.
+    void reachedEndOfTopic();
     inline CommandSubscribe_SubType getSubType();
     inline CommandSubscribe_InitialPosition getInitialPosition();
 
@@ -185,6 +188,10 @@ class ConsumerImpl : public ConsumerImplBase {
 
    private:
     std::atomic_bool waitingForZeroQueueSizeMessage;
+    // Set when the broker sends CommandReachedEndOfTopic and cleared again on each new broker
+    // session (termination does not cancel redelivery of unacked messages); a drained receive
+    // then yields ResultTopicTerminated instead of parking forever.
+    std::atomic_bool hasReachedEndOfTopic_{false};
     std::shared_ptr<ConsumerImpl> get_shared_this_ptr();
     bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData,
                                    const proto::MessageMetadata& metadata, SharedBuffer& payload,

lib/st/MessageCore.cc

@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include <pulsar/st/detail/MessageCore.h>
+
+#include "MessageImpl.h"
+
+namespace pulsar::st::detail {
+
+// Thin forwarders to the hidden MessageImpl (see ProducerCore.cc for the same pattern).
+std::span<const std::byte> MessageCore::data() const { return impl_->data(); }
+MessageId MessageCore::id() const { return impl_->id(); }
+std::optional<std::string_view> MessageCore::key() const { return impl_->key(); }
+const Properties& MessageCore::properties() const { return impl_->properties(); }
+Timestamp MessageCore::publishTime() const { return impl_->publishTime(); }
+std::optional<Timestamp> MessageCore::eventTime() const { return impl_->eventTime(); }
+int64_t MessageCore::sequenceId() const { return impl_->sequenceId(); }
+std::optional<std::string_view> MessageCore::producerName() const { return impl_->producerName(); }
+std::string_view MessageCore::topic() const { return impl_->topic(); }
+int MessageCore::redeliveryCount() const { return impl_->redeliveryCount(); }
+std::optional<std::string_view> MessageCore::replicatedFrom() const { return impl_->replicatedFrom(); }
+
+}  // namespace pulsar::st::detail

lib/st/MessageImpl.h

@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <pulsar/Message.h>
+#include <pulsar/st/MessageId.h>
+#include <pulsar/st/detail/MessageCore.h>
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <optional>
+#include <span>
+#include <string>
+#include <string_view>
+#include <utility>
+
+namespace pulsar::st {
+
+/**
+ * INTERNAL — the received message behind `detail::MessageCore`.
+ *
+ * A thin view over a classic `pulsar::Message` (which owns the payload and metadata)
+ * plus the segment-qualified `pulsar::st::MessageId` minted on the receive path. An
+ * optional `topicOverride` carries the scalable topic identity in namespace mode
+ * (a plain segment consumer reports the segment backing topic otherwise).
+ */
+class MessageImpl {
+   public:
+    MessageImpl(pulsar::Message message, MessageId id,
+                std::optional<std::string> topicOverride = std::nullopt)
+        : classic_(std::move(message)), id_(std::move(id)), topicOverride_(std::move(topicOverride)) {}
+
+    std::span<const std::byte> data() const {
+        return {static_cast<const std::byte*>(classic_.getData()), classic_.getLength()};
+    }
+    const MessageId& id() const { return id_; }
+    std::optional<std::string_view> key() const {
+        if (!classic_.hasPartitionKey()) return std::nullopt;
+        return std::string_view(classic_.getPartitionKey());
+    }
+    const Properties& properties() const { return classic_.getProperties(); }
+    Timestamp publishTime() const { return fromMillis(classic_.getPublishTimestamp()); }
+    std::optional<Timestamp> eventTime() const {
+        const uint64_t millis = classic_.getEventTimestamp();
+        return millis != 0 ? std::optional<Timestamp>(fromMillis(millis)) : std::nullopt;
+    }
+    // The classic public Message API does not expose the message's sequence id; populating it
+    // would require reaching into pulsar::MessageImpl's metadata, i.e. touching the classic API.
+    // TODO: revisit when the Stream consumer needs it (a classic Message::getSequenceId() accessor).
+    int64_t sequenceId() const { return -1; }
+    std::optional<std::string_view> producerName() const {
+        const std::string& name = classic_.getProducerName();
+        return name.empty() ? std::nullopt : std::optional<std::string_view>(name);
+    }
+    std::string_view topic() const {
+        return topicOverride_ ? std::string_view(*topicOverride_) : std::string_view(classic_.getTopicName());
+    }
+    int redeliveryCount() const { return classic_.getRedeliveryCount(); }
+    std::optional<std::string_view> replicatedFrom() const {
+        const std::optional<const std::string*> from = classic_.getReplicatedFrom();
+        if (!from || *from == nullptr) return std::nullopt;
+        return std::string_view(**from);
+    }
+
+   private:
+    static Timestamp fromMillis(uint64_t millis) {
+        return Timestamp(std::chrono::milliseconds(static_cast<std::int64_t>(millis)));
+    }
+
+    pulsar::Message classic_;
+    MessageId id_;
+    std::optional<std::string> topicOverride_;
+};
+
+}  // namespace pulsar::st