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

include/pulsar/st/detail/StreamConsumerCore.h

@@ -34,6 +34,7 @@ namespace pulsar::st {
 class StreamConsumerImpl;
 using StreamConsumerImplPtr = std::shared_ptr<StreamConsumerImpl>;
 class Transaction;
+class ClientImpl;  // lib/st — mints consumer cores from subscribeStreamAsync
 
 namespace detail {
 
@@ -62,6 +63,7 @@ class PULSAR_PUBLIC StreamConsumerCore {
 
    private:
     friend class ClientCore;
+    friend class ::pulsar::st::ClientImpl;
     explicit StreamConsumerCore(StreamConsumerImplPtr impl) : impl_(std::move(impl)) {}
 
     StreamConsumerImplPtr impl_;

lib/ClientConnection.cc

@@ -1018,6 +1018,14 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {
                     handleScalableTopicUpdate(incomingCmd.scalabletopicupdate());
                     break;
 
+                case BaseCommand::SCALABLE_TOPIC_SUBSCRIBE_RESPONSE:
+                    handleScalableTopicSubscribeResponse(incomingCmd.scalabletopicsubscriberesponse());
+                    break;
+
+                case BaseCommand::SCALABLE_TOPIC_ASSIGNMENT_UPDATE:
+                    handleScalableTopicAssignmentUpdate(incomingCmd.scalabletopicassignmentupdate());
+                    break;
+
                 case BaseCommand::REACHED_END_OF_TOPIC:
                     handleReachedEndOfTopic(incomingCmd.reachedendoftopic());
                     break;
@@ -1302,6 +1310,8 @@ const std::future<void>& ClientConnection::close(Error&& error, bool switchClust
     auto pendingGetNamespaceTopicsRequests = std::move(pendingGetNamespaceTopicsRequests_);
     auto pendingGetSchemaRequests = std::move(pendingGetSchemaRequests_);
     auto scalableTopicSessions = std::move(scalableTopicSessions_);
+    auto scalableConsumerSessions = std::move(scalableConsumerSessions_);
+    auto pendingScalableSubscribeRequests = std::move(pendingScalableSubscribeRequests_);
 
     numOfPendingLookupRequest_ = 0;
 
@@ -1381,6 +1391,12 @@ const std::future<void>& ClientConnection::close(Error&& error, bool switchClust
     for (auto& kv : scalableTopicSessions) {
         kv.second(error.result, nullptr);
     }
+    for (auto& kv : scalableConsumerSessions) {
+        kv.second(error.result, nullptr);
+    }
+    for (auto& kv : pendingScalableSubscribeRequests) {
+        kv.second(error.result, nullptr);
+    }
     for (auto& kv : pendingConsumerStatsMap) {
         LOG_ERROR(cnxString() << " Closing Client Connection, please try again later");
         kv.second.setFailed(result);
@@ -1434,6 +1450,75 @@ void ClientConnection::removeScalableTopicSession(uint64_t sessionId) {
     scalableTopicSessions_.erase(sessionId);
 }
 
+bool ClientConnection::registerScalableConsumerSession(uint64_t consumerId,
+                                                       ScalableConsumerAssignmentListener listener) {
+    Lock lock(mutex_);
+    if (isClosed()) {
+        return false;
+    }
+    scalableConsumerSessions_[consumerId] = std::move(listener);
+    return true;
+}
+
+void ClientConnection::removeScalableConsumerSession(uint64_t consumerId) {
+    Lock lock(mutex_);
+    scalableConsumerSessions_.erase(consumerId);
+}
+
+bool ClientConnection::addScalableSubscribeRequest(uint64_t requestId,
+                                                   ScalableSubscribeResponseCallback callback) {
+    Lock lock(mutex_);
+    if (isClosed()) {
+        return false;
+    }
+    pendingScalableSubscribeRequests_[requestId] = std::move(callback);
+    return true;
+}
+
+void ClientConnection::removeScalableSubscribeRequest(uint64_t requestId) {
+    Lock lock(mutex_);
+    pendingScalableSubscribeRequests_.erase(requestId);
+}
+
+void ClientConnection::handleScalableTopicSubscribeResponse(
+    const proto::CommandScalableTopicSubscribeResponse& response) {
+    ScalableSubscribeResponseCallback callback;
+    {
+        Lock lock(mutex_);
+        auto it = pendingScalableSubscribeRequests_.find(response.request_id());
+        if (it != pendingScalableSubscribeRequests_.end()) {
+            callback = std::move(it->second);
+            pendingScalableSubscribeRequests_.erase(it);
+        }
+    }
+    if (callback) {
+        callback(ResultOk, &response);
+    } else {
+        LOG_WARN(cnxString() << "Received SCALABLE_TOPIC_SUBSCRIBE_RESPONSE for unknown request "
+                             << response.request_id());
+    }
+}
+
+void ClientConnection::handleScalableTopicAssignmentUpdate(
+    const proto::CommandScalableTopicAssignmentUpdate& update) {
+    ScalableConsumerAssignmentListener listener;
+    {
+        Lock lock(mutex_);
+        auto it = scalableConsumerSessions_.find(update.consumer_id());
+        if (it != scalableConsumerSessions_.end()) {
+            listener = it->second;
+        }
+    }
+    if (listener) {
+        listener(ResultOk, &update);
+    } else {
+        // A push may race with a just-closed session; drop it rather than
+        // treating it as a protocol violation.
+        LOG_WARN(cnxString() << "Received SCALABLE_TOPIC_ASSIGNMENT_UPDATE for unknown consumer "
+                             << update.consumer_id());
+    }
+}
+
 void ClientConnection::handleScalableTopicUpdate(const proto::CommandScalableTopicUpdate& update) {
     ScalableTopicUpdateListener listener;
     {

lib/ClientConnection.h

@@ -106,6 +106,8 @@ class CommandLookupTopicResponse;
 class CommandPartitionedTopicMetadataResponse;
 class CommandProducerSuccess;
 class CommandReachedEndOfTopic;
+class CommandScalableTopicAssignmentUpdate;
+class CommandScalableTopicSubscribeResponse;
 class CommandScalableTopicUpdate;
 class CommandSendReceipt;
 class CommandSendError;
@@ -203,6 +205,34 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
     bool registerScalableTopicSession(uint64_t sessionId, ScalableTopicUpdateListener listener);
     void removeScalableTopicSession(uint64_t sessionId);
 
+    // Scalable topics (pulsar::st): a stream/checkpoint consumer session registered with
+    // the controller through this connection. The listener is invoked with
+    // (ResultOk, &update) for every CommandScalableTopicAssignmentUpdate whose
+    // consumer_id matches, and with (error, nullptr) once when the connection closes,
+    // after which the registration is gone.
+    typedef std::function<void(Result, const proto::CommandScalableTopicAssignmentUpdate*)>
+        ScalableConsumerAssignmentListener;
+    // One-shot callback for a CommandScalableTopicSubscribeResponse, correlated by
+    // request_id: (ResultOk, &response) when the response arrives — the response may
+    // itself carry a broker error — or (error, nullptr) once if the connection closes
+    // first. Removed from the registry when fired.
+    typedef std::function<void(Result, const proto::CommandScalableTopicSubscribeResponse*)>
+        ScalableSubscribeResponseCallback;
+
+    /**
+     * Register a consumer session for pushed assignment updates. Returns false
+     * (without registering) if the connection is already closed.
+     */
+    bool registerScalableConsumerSession(uint64_t consumerId, ScalableConsumerAssignmentListener listener);
+    void removeScalableConsumerSession(uint64_t consumerId);
+
+    /**
+     * Register a one-shot callback for the subscribe response with this request id.
+     * Returns false (without registering) if the connection is already closed.
+     */
+    bool addScalableSubscribeRequest(uint64_t requestId, ScalableSubscribeResponseCallback callback);
+    void removeScalableSubscribeRequest(uint64_t requestId);
+
     /** Whether the broker advertised scalable-topics support on CONNECTED. */
     bool supportsScalableTopics() const { return supportsScalableTopics_.load(std::memory_order_acquire); }
 
@@ -377,6 +407,10 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
     // Scalable topics: DAG-watch sessions by client-assigned session id.
     typedef std::map<uint64_t, ScalableTopicUpdateListener> ScalableTopicSessionsMap;
     ScalableTopicSessionsMap scalableTopicSessions_;
+    typedef std::map<uint64_t, ScalableConsumerAssignmentListener> ScalableConsumerSessionsMap;
+    ScalableConsumerSessionsMap scalableConsumerSessions_;
+    typedef std::map<uint64_t, ScalableSubscribeResponseCallback> ScalableSubscribeRequestsMap;
+    ScalableSubscribeRequestsMap pendingScalableSubscribeRequests_;
     std::atomic<bool> supportsScalableTopics_{false};
 
     typedef std::map<uint64_t, Promise<Result, BrokerConsumerStatsImpl>> PendingConsumerStatsMap;
@@ -462,6 +496,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
     void handleGetSchemaResponse(const proto::CommandGetSchemaResponse&);
     void handleAckResponse(const proto::CommandAckResponse&);
     void handleScalableTopicUpdate(const proto::CommandScalableTopicUpdate&);
+    void handleScalableTopicSubscribeResponse(const proto::CommandScalableTopicSubscribeResponse&);
+    void handleScalableTopicAssignmentUpdate(const proto::CommandScalableTopicAssignmentUpdate&);
     optional<std::string> getAssignedBrokerServiceUrl(const proto::CommandCloseProducer&);
     optional<std::string> getAssignedBrokerServiceUrl(const proto::CommandCloseConsumer&);
     std::string getMigratedBrokerServiceUrl(const proto::CommandTopicMigrated&);

lib/ClientImpl.cc

@@ -625,9 +625,13 @@ void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::s
             lock.unlock();
             callback(Error{ResultInvalidTopicName, ""});
             return;
-        } else if (conf.isReadCompacted() && (topicName->getDomain().compare("persistent") != 0 ||
-                                              (conf.getConsumerType() != ConsumerExclusive &&
-                                               conf.getConsumerType() != ConsumerFailover))) {
+        } else if (conf.isReadCompacted() &&
+                   // Segment backing topics are persistent in all but the scheme, so the
+                   // scalable-topics consumers may read them compacted too.
+                   ((topicName->getDomain().compare("persistent") != 0 &&
+                     !(allowSegmentTopic && topicName->isSegment())) ||
+                    (conf.getConsumerType() != ConsumerExclusive &&
+                     conf.getConsumerType() != ConsumerFailover))) {
             lock.unlock();
             callback(Error{ResultInvalidConfiguration, ""});
             return;

lib/Commands.cc

@@ -554,6 +554,22 @@ SharedBuffer Commands::newCloseConsumer(uint64_t consumerId, uint64_t requestId)
     return writeMessageWithSize(cmd);
 }
 
+SharedBuffer Commands::newScalableTopicSubscribe(uint64_t requestId, const std::string& topic,
+                                                 const std::string& subscription,
+                                                 const std::string& consumerName, uint64_t consumerId,
+                                                 ScalableConsumerType consumerType) {
+    BaseCommand cmd;
+    cmd.set_type(BaseCommand::SCALABLE_TOPIC_SUBSCRIBE);
+    proto::CommandScalableTopicSubscribe* subscribe = cmd.mutable_scalabletopicsubscribe();
+    subscribe->set_request_id(requestId);
+    subscribe->set_topic(topic);
+    subscribe->set_subscription(subscription);
+    subscribe->set_consumer_name(consumerName);
+    subscribe->set_consumer_id(consumerId);
+    subscribe->set_consumer_type(static_cast<proto::ScalableConsumerType>(consumerType));
+    return writeMessageWithSize(cmd);
+}
+
 SharedBuffer Commands::newScalableTopicLookup(uint64_t sessionId, const std::string& topic,
                                               bool createIfMissing) {
     BaseCommand cmd;

lib/Commands.h

@@ -150,6 +150,10 @@ class Commands {
     // Scalable topics (pulsar::st): open/close a DAG-watch session. The broker
     // answers (and later pushes) CommandScalableTopicUpdate correlated by the
     // client-assigned sessionId.
+    static SharedBuffer newScalableTopicSubscribe(uint64_t requestId, const std::string& topic,
+                                                  const std::string& subscription,
+                                                  const std::string& consumerName, uint64_t consumerId,
+                                                  ScalableConsumerType consumerType);
     static SharedBuffer newScalableTopicLookup(uint64_t sessionId, const std::string& topic,
                                                bool createIfMissing);
     static SharedBuffer newScalableTopicClose(uint64_t sessionId);

lib/ProtoApiEnums.h

@@ -158,4 +158,8 @@ constexpr BaseCommand_Type BaseCommand_Type_WATCH_TOPIC_LIST_SUCCESS = 65;
 constexpr BaseCommand_Type BaseCommand_Type_WATCH_TOPIC_UPDATE = 66;
 constexpr BaseCommand_Type BaseCommand_Type_WATCH_TOPIC_LIST_CLOSE = 67;
 
+using ScalableConsumerType = int;
+constexpr ScalableConsumerType ScalableConsumerType_STREAM = 0;
+constexpr ScalableConsumerType ScalableConsumerType_CHECKPOINT = 1;
+
 }  // namespace pulsar