Scalable Topics: pulsar::st queue consumer — per-segment fan-in over a mux receive queue

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