From a5e2f123a28309c068586f6b18a9ec48115f7d2c Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 09:31:02 -0700 Subject: [PATCH 1/5] =?UTF-8?q?st:=20consumer=20assignment=20session=20?= =?UTF-8?q?=E2=80=94=20controller=20registration=20for=20stream=20consumer?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stream consumer is told what to consume, unlike the queue consumer which watches the DAG and attaches to every segment itself: ordering across splits and merges is enforced broker-side by the SubscriptionCoordinator, which withholds a child segment from the assignment until every parent has drained. The client's job is to register with the controller and subscribe to exactly the assigned segments. This adds the controller session (a port of the Java v5 ScalableConsumerClient): - Commands::newScalableTopicSubscribe and the ScalableConsumerType constants. - ClientConnection: a consumer-session registry dispatching pushed CommandScalableTopicAssignmentUpdate by consumer id, one-shot callbacks for CommandScalableTopicSubscribeResponse correlated by request id, and close notification for both — the same idiom as the DAG-watch session registry. - ConsumerAssignmentSession: resolve 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 the subscribe), register + subscribe, apply epoch-gated assignments, replay the current assignment on listener registration, reconnect with backoff after the initial assignment and fail fast before it. close() drops only the local registration; the broker reaps the registration through its grace timer (Java parity). --- lib/ClientConnection.cc | 85 ++++++ lib/ClientConnection.h | 36 +++ lib/Commands.cc | 16 + lib/Commands.h | 4 + lib/ProtoApiEnums.h | 4 + lib/st/ConsumerAssignmentSession.cc | 340 ++++++++++++++++++++++ lib/st/ConsumerAssignmentSession.h | 139 +++++++++ tests/st/ConsumerAssignmentSessionTest.cc | 81 ++++++ 8 files changed, 705 insertions(+) create mode 100644 lib/st/ConsumerAssignmentSession.cc create mode 100644 lib/st/ConsumerAssignmentSession.h create mode 100644 tests/st/ConsumerAssignmentSessionTest.cc diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc index 39442ca5..1bc1738f 100644 --- a/lib/ClientConnection.cc +++ b/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& 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& 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; { diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h index 05f8578e..e67d6a4a 100644 --- a/lib/ClientConnection.h +++ b/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 + 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 + 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 ScalableTopicSessionsMap; ScalableTopicSessionsMap scalableTopicSessions_; + typedef std::map ScalableConsumerSessionsMap; + ScalableConsumerSessionsMap scalableConsumerSessions_; + typedef std::map ScalableSubscribeRequestsMap; + ScalableSubscribeRequestsMap pendingScalableSubscribeRequests_; std::atomic supportsScalableTopics_{false}; typedef std::map> PendingConsumerStatsMap; @@ -462,6 +496,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this getAssignedBrokerServiceUrl(const proto::CommandCloseProducer&); optional getAssignedBrokerServiceUrl(const proto::CommandCloseConsumer&); std::string getMigratedBrokerServiceUrl(const proto::CommandTopicMigrated&); diff --git a/lib/Commands.cc b/lib/Commands.cc index 9b800874..ea609ee7 100644 --- a/lib/Commands.cc +++ b/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(consumerType)); + return writeMessageWithSize(cmd); +} + SharedBuffer Commands::newScalableTopicLookup(uint64_t sessionId, const std::string& topic, bool createIfMissing) { BaseCommand cmd; diff --git a/lib/Commands.h b/lib/Commands.h index 622f95d7..459e03b4 100644 --- a/lib/Commands.h +++ b/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); diff --git a/lib/ProtoApiEnums.h b/lib/ProtoApiEnums.h index 5f1876bd..a435c1e1 100644 --- a/lib/ProtoApiEnums.h +++ b/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 diff --git a/lib/st/ConsumerAssignmentSession.cc b/lib/st/ConsumerAssignmentSession.cc new file mode 100644 index 00000000..b90f5301 --- /dev/null +++ b/lib/st/ConsumerAssignmentSession.cc @@ -0,0 +1,340 @@ +/** + * 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 "ConsumerAssignmentSession.h" + +#include +#include + +#include "DagWatchSession.h" +#include "PulsarApi.pb.h" +#include "lib/Commands.h" +#include "lib/ExecutorService.h" +#include "lib/LogUtils.h" + +DECLARE_LOG_OBJECT() + +namespace pulsar::st { + +namespace { + +Result toSubscribeResult(pulsar::proto::ServerError error) { + switch (error) { + case pulsar::proto::TopicNotFound: + return ResultTopicNotFound; + case pulsar::proto::ConsumerBusy: + return ResultConsumerBusy; + default: + return ResultUnknownError; + } +} + +ConsumerAssignment fromProto(const pulsar::proto::ScalableConsumerAssignment& proto) { + ConsumerAssignment assignment; + assignment.layoutEpoch = proto.layout_epoch(); + assignment.segments.reserve(proto.segments_size()); + for (int i = 0; i < proto.segments_size(); i++) { + const auto& s = proto.segments(i); + AssignedSegment assigned; + assigned.segment.segmentId = s.segment_id(); + assigned.segment.range = HashRange{s.hash_start(), s.hash_end()}; + assigned.segment.segmentTopicName = s.segment_topic(); + assigned.ownedBucketRanges.reserve(s.bucket_ranges_size()); + for (int j = 0; j < s.bucket_ranges_size(); j++) { + const auto& range = s.bucket_ranges(j); + assigned.ownedBucketRanges.push_back(HashRange{static_cast(range.start()), + static_cast(range.end())}); + } + assignment.segments.push_back(std::move(assigned)); + } + return assignment; +} + +} // namespace + +ConsumerAssignmentSession::ConsumerAssignmentSession(pulsar::ClientImplPtr client, std::string topic, + std::string subscription, std::string consumerName, + pulsar::ScalableConsumerType consumerType) + : client_(std::move(client)), + topic_(std::move(topic)), + subscription_(std::move(subscription)), + consumerName_(std::move(consumerName)), + consumerType_(consumerType), + consumerId_(client_->newConsumerId()), + backoff_(std::chrono::milliseconds(100), std::chrono::seconds(30), std::chrono::milliseconds(0)), + reconnectTimer_(client_->getIOExecutorProvider()->get()->createDeadlineTimer()) {} + +Future> ConsumerAssignmentSession::start() { + detail::Promise attempt; + auto self = shared_from_this(); + attempt.getFuture().addListener([self](const Expected& result) { + if (result) { + // Mark before applying so a connection drop racing the response reconnects + // instead of failing the (already satisfied) subscribe. + self->sawInitialAssignment_.store(true); + self->handleAssignmentReceived(*result); + self->initialAssignmentPromise_.setValue(result->segments); + } else { + self->initialAssignmentPromise_.setError(result.error()); + } + }); + connectAndSubscribe(attempt); + return initialAssignmentPromise_.getFuture(); +} + +void ConsumerAssignmentSession::connectAndSubscribe(detail::Promise promise) { + // Resolve the controller leader through a one-shot DAG-watch lookup: scalable + // topic URIs are not resolvable through the classic lookup service, and the + // controller pushes assignment updates itself, so no long-lived layout watch is + // needed. The watch is closed as soon as the layout arrives. + auto self = shared_from_this(); + auto watch = std::make_shared(client_, topic_, /*createIfMissing*/ true); + watch->start().addListener([self, watch, promise](const Expected& result) { + watch->close(); + if (!result) { + promise.setError(result.error()); + return; + } + if (self->closed_.load()) { + promise.setError(Error{ResultAlreadyClosed, "consumer session closed"}); + return; + } + const bool useTls = self->client_->getServiceInfo().useTls(); + const auto& controllerUrl = + useTls ? result->controllerBrokerUrlTls() : result->controllerBrokerUrl(); + // Behind a proxy the controller's advertised address is not directly reachable, + // and before leader election completes there is no address at all: in both + // cases connect through the regular lookup path — any broker forwards the + // subscribe to the controller and relays assignment updates back. + const bool useDirect = + controllerUrl.has_value() && !controllerUrl->empty() && + self->client_->getClientConfig().getProxyServiceUrl().empty(); + auto connectionFuture = + useDirect ? self->client_->connect("", *controllerUrl, static_cast(self->consumerId_)) + : self->client_->getConnection("", DagWatchSession::lookupCompatibleTopic(self->topic_), + static_cast(self->consumerId_)); + connectionFuture.addListener( + [self, promise](pulsar::Result result, const pulsar::ClientConnectionPtr& cnx) { + if (result == pulsar::ResultOk) { + self->subscribeOn(cnx, promise); + } else { + promise.setError(Error{result, "failed to connect to the scalable-topics controller"}); + } + }); + }); +} + +void ConsumerAssignmentSession::subscribeOn(const pulsar::ClientConnectionPtr& cnx, + detail::Promise promise) { + if (closed_.load()) { + promise.setError(Error{ResultAlreadyClosed, "consumer session closed"}); + return; + } + if (!cnx->supportsScalableTopics()) { + promise.setError(Error{ResultUnsupportedVersionError, "the broker does not support scalable topics"}); + return; + } + { + std::lock_guard lock(mutex_); + cnx_ = cnx; + } + std::weak_ptr weakSelf = weak_from_this(); + bool registered = cnx->registerScalableConsumerSession( + consumerId_, + [weakSelf](pulsar::Result result, const pulsar::proto::CommandScalableTopicAssignmentUpdate* update) { + if (auto self = weakSelf.lock()) self->handleSessionEvent(result, update); + }); + if (!registered) { + // The connection closed between acquisition and registration. + promise.setError(Error{ResultNotConnected, "connection closed before the consumer registration"}); + return; + } + const std::uint64_t requestId = client_->newRequestId(); + bool added = cnx->addScalableSubscribeRequest( + requestId, + [promise](pulsar::Result result, const pulsar::proto::CommandScalableTopicSubscribeResponse* response) { + if (result != pulsar::ResultOk) { + promise.setError(Error{result, "connection closed before the subscribe response"}); + return; + } + if (response->has_error()) { + promise.setError(Error{toSubscribeResult(response->error()), + response->has_message() ? response->message() + : "scalable-topic subscribe failed"}); + return; + } + if (!response->has_assignment()) { + promise.setError( + Error{ResultUnknownError, "subscribe response carried neither assignment nor error"}); + return; + } + promise.setValue(fromProto(response->assignment())); + }); + if (!added) { + promise.setError(Error{ResultNotConnected, "connection closed before the subscribe request"}); + return; + } + // A failed write closes the connection, which fails the pending request and the + // session registration through the close notification — no separate error path. + cnx->sendCommand(Commands::newScalableTopicSubscribe(requestId, topic_, subscription_, consumerName_, + consumerId_, consumerType_)); +} + +void ConsumerAssignmentSession::handleSessionEvent( + pulsar::Result result, const pulsar::proto::CommandScalableTopicAssignmentUpdate* update) { + if (closed_.load()) { + return; + } + if (result != pulsar::ResultOk) { + handleConnectionClosed(); + return; + } + handleAssignmentReceived(fromProto(update->assignment())); +} + +void ConsumerAssignmentSession::handleAssignmentReceived(const ConsumerAssignment& assignment) { + std::vector newSegments; + std::vector oldSegments; + AssignmentChangeListener listener; + { + std::lock_guard lock(mutex_); + const auto epoch = static_cast(assignment.layoutEpoch); + if (epoch < currentEpoch_) { + LOG_INFO("[" << topic_ << "] consumer " << consumerId_ << ": ignoring stale assignment (epoch " + << epoch << " < " << currentEpoch_ << ")"); + return; + } + oldSegments = std::move(currentAssignment_); + currentAssignment_ = assignment.segments; + currentEpoch_ = epoch; + newSegments = currentAssignment_; + listener = listener_; + } + LOG_INFO("[" << topic_ << "] consumer " << consumerId_ << ": assignment updated (epoch " + << assignment.layoutEpoch << ", " << newSegments.size() << " segments)"); + if (listener) { + listener(newSegments, oldSegments); + } +} + +void ConsumerAssignmentSession::handleConnectionClosed() { + LOG_WARN("[" << topic_ << "] consumer " << consumerId_ << ": assignment session connection closed"); + { + std::lock_guard lock(mutex_); + cnx_.reset(); + } + if (closed_.load()) { + return; + } + if (!sawInitialAssignment_.load()) { + // The initial subscribe never completed: surface the failure to the caller + // rather than retrying silently. (First-writer-wins, so this cannot override + // a response that raced the close.) + initialAssignmentPromise_.setError( + Error{ResultNotConnected, "connection closed before the initial assignment arrived"}); + return; + } + scheduleReconnect(); +} + +void ConsumerAssignmentSession::scheduleReconnect() { + if (closed_.load()) { + return; + } + auto delay = backoff_.next(); + LOG_INFO("[" << topic_ << "] consumer " << consumerId_ << ": reconnecting the assignment session in " + << std::chrono::duration_cast(delay).count() << " ms"); + std::weak_ptr weakSelf = shared_from_this(); + reconnectTimer_->expires_from_now(delay); + reconnectTimer_->async_wait([weakSelf](const ASIO_ERROR& error) { + auto self = weakSelf.lock(); + if (self && !error) { + self->reconnect(); + } + }); +} + +void ConsumerAssignmentSession::reconnect() { + if (closed_.load()) { + return; + } + detail::Promise attempt; + auto self = shared_from_this(); + attempt.getFuture().addListener([self](const Expected& result) { + if (self->closed_.load()) { + return; + } + if (result) { + // Feed the response through the standard update path so the listener gets + // the diff. Within the controller's grace period this is a no-op (same + // segments); past it the controller has rebalanced and the listener + // attaches/detaches accordingly. + self->backoff_.reset(); + self->handleAssignmentReceived(*result); + } else { + LOG_WARN("[" << self->topic_ << "] consumer " << self->consumerId_ + << ": assignment session reconnect failed (" << result.error() << "); will retry"); + self->scheduleReconnect(); + } + }); + connectAndSubscribe(attempt); +} + +std::vector ConsumerAssignmentSession::currentAssignment() const { + std::lock_guard lock(mutex_); + return currentAssignment_; +} + +void ConsumerAssignmentSession::setListener(AssignmentChangeListener listener) { + std::vector current; + AssignmentChangeListener toReplay; + { + std::lock_guard lock(mutex_); + listener_ = std::move(listener); + current = currentAssignment_; + toReplay = listener_; + } + // Replay the current assignment: an update that raced the registration would + // otherwise be lost — there is no periodic refresh to recover it. Appliers are + // idempotent, so a redundant replay is a no-op. + if (toReplay) { + toReplay(current, current); + } +} + +void ConsumerAssignmentSession::close() { + if (closed_.exchange(true)) { + return; + } + ASIO_ERROR ignored; + reconnectTimer_->cancel(ignored); + + pulsar::ClientConnectionPtr cnx; + { + std::lock_guard lock(mutex_); + cnx = cnx_.lock(); + cnx_.reset(); + } + if (cnx) { + // No wire command: the broker reaps the registration through its grace timer + // on disconnect (Java parity). + cnx->removeScalableConsumerSession(consumerId_); + } +} + +} // namespace pulsar::st diff --git a/lib/st/ConsumerAssignmentSession.h b/lib/st/ConsumerAssignmentSession.h new file mode 100644 index 00000000..d91c7085 --- /dev/null +++ b/lib/st/ConsumerAssignmentSession.h @@ -0,0 +1,139 @@ +/** + * 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 + +#include +#include +#include +#include +#include +#include +#include + +#include "SegmentLayout.h" +#include "lib/Backoff.h" +#include "lib/ClientConnection.h" +#include "lib/ClientImpl.h" +#include "lib/ProtoApiEnums.h" + +namespace pulsar::st { + +/** One segment assigned to this consumer by the controller. */ +struct AssignedSegment { + Segment segment; + /** + * PIP-486: the entry-bucket hash ranges this consumer owns within the segment. + * Empty means the consumer owns the whole segment and subscribes Exclusive; + * non-empty means the segment is shared by bucket (Key_Shared STICKY). + */ + std::vector ownedBucketRanges; +}; + +/** The controller's assignment of segments to this consumer at one layout epoch. */ +struct ConsumerAssignment { + std::uint64_t layoutEpoch = 0; + std::vector segments; +}; + +/** + * The controller-registration session of one stream/checkpoint consumer, ported from + * the Java v5 client's ScalableConsumerClient so the two clients behave identically. + * + * Unlike the queue consumer — which watches the DAG itself and attaches to every + * segment — a stream consumer is told what to consume: start() resolves the + * controller leader's URL through a one-shot DAG-watch lookup, connects to it, + * registers this consumer (CommandScalableTopicSubscribe) and completes with the + * initial segment assignment. The controller pushes a new assignment + * (CommandScalableTopicAssignmentUpdate) after every rebalance — a peer joining or + * leaving the subscription, or a segment split/merge whose parents have drained — + * and each accepted (non-stale by layout epoch) assignment is reported to the + * listener. When the connection drops after the initial assignment the session + * reconnects with exponential backoff and re-subscribes (within the controller's + * grace period that returns the same assignment); if it drops before the first + * assignment, start()'s future fails instead. close() only removes the local + * registration — the broker reaps the registration through its grace timer. + */ +class ConsumerAssignmentSession : public std::enable_shared_from_this { + public: + /** Invoked for every accepted assignment after the initial one (and on setListener replay). */ + using AssignmentChangeListener = std::function& newSegments, + const std::vector& oldSegments)>; + + ConsumerAssignmentSession(pulsar::ClientImplPtr client, std::string topic, std::string subscription, + std::string consumerName, pulsar::ScalableConsumerType consumerType); + + /** + * Start the session. May be called once. + * @return a future completing with the initial assignment's segments, or the failure. + */ + Future> start(); + + /** Snapshot of the most recent assignment (empty before the first one). */ + std::vector currentAssignment() const; + + /** + * Register the listener notified on every accepted assignment update. Replays the + * current assignment immediately (newSegments == oldSegments) so an update that + * raced the registration is not lost; appliers must be idempotent. + */ + void setListener(AssignmentChangeListener listener); + + /** Close the session: stop reconnecting and drop the local registration. Idempotent. */ + void close(); + + std::uint64_t consumerId() const { return consumerId_; } + + private: + // One connect-register-subscribe attempt; completes the promise with the + // controller's assignment or the first failure. Used by start() and reconnect(). + void connectAndSubscribe(detail::Promise promise); + void subscribeOn(const pulsar::ClientConnectionPtr& cnx, detail::Promise promise); + void handleSessionEvent(pulsar::Result result, + const pulsar::proto::CommandScalableTopicAssignmentUpdate* update); + // Epoch-gated apply + listener notification; used for the initial assignment, + // pushed updates, and reconnect responses alike. + void handleAssignmentReceived(const ConsumerAssignment& assignment); + void handleConnectionClosed(); + void scheduleReconnect(); + void reconnect(); + + pulsar::ClientImplPtr client_; + const std::string topic_; + const std::string subscription_; + const std::string consumerName_; + const pulsar::ScalableConsumerType consumerType_; + const std::uint64_t consumerId_; + pulsar::Backoff backoff_; + DeadlineTimerPtr reconnectTimer_; // global-scope alias from AsioTimer.h + detail::Promise> initialAssignmentPromise_; + std::atomic sawInitialAssignment_{false}; + std::atomic closed_{false}; + + mutable std::mutex mutex_; + std::vector currentAssignment_; // guarded by mutex_ + std::int64_t currentEpoch_ = -1; // guarded by mutex_ + AssignmentChangeListener listener_; // guarded by mutex_ + pulsar::ClientConnectionWeakPtr cnx_; // guarded by mutex_ +}; + +using ConsumerAssignmentSessionPtr = std::shared_ptr; + +} // namespace pulsar::st diff --git a/tests/st/ConsumerAssignmentSessionTest.cc b/tests/st/ConsumerAssignmentSessionTest.cc new file mode 100644 index 00000000..ddbf9cef --- /dev/null +++ b/tests/st/ConsumerAssignmentSessionTest.cc @@ -0,0 +1,81 @@ +/** + * 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 +#include + +#include +#include + +#include "lib/ClientImpl.h" +#include "lib/st/ConsumerAssignmentSession.h" + +// Broker-free tests for the consumer assignment session: the lifecycle paths that +// must be safe without any connection, and the listener replay contract. The full +// session protocol (controller lookup, subscribe, assignment updates, reconnect) is +// exercised end-to-end against a real broker by the stream-consumer integration +// tests. + +using namespace pulsar::st; + +TEST(ConsumerAssignmentSessionTest, testSessionLifecycleWithoutConnection) { + auto classic = + std::make_shared("pulsar://localhost:6650", pulsar::ClientConfiguration{}); + classic->initialize(); + + auto session = std::make_shared( + classic, "topic://public/default/orders", "sub", "consumer-1", pulsar::ScalableConsumerType_STREAM); + + // Fresh session: empty assignment, a real consumer id, distinct per session. + ASSERT_TRUE(session->currentAssignment().empty()); + auto other = std::make_shared( + classic, "topic://public/default/other", "sub", "consumer-2", pulsar::ScalableConsumerType_STREAM); + ASSERT_NE(session->consumerId(), other->consumerId()); + + // Closing before start (and closing twice) must be safe. + session->close(); + session->close(); + other->close(); + + classic->shutdown(); +} + +TEST(ConsumerAssignmentSessionTest, testSetListenerReplaysCurrentAssignment) { + auto classic = + std::make_shared("pulsar://localhost:6650", pulsar::ClientConfiguration{}); + classic->initialize(); + + auto session = std::make_shared( + classic, "topic://public/default/orders", "sub", "consumer-1", pulsar::ScalableConsumerType_STREAM); + + // Before any assignment the replay still fires (with empty old == new), so an + // applier registered late cannot miss the registration race window. + int calls = 0; + std::vector seenNew; + session->setListener([&](const std::vector& newSegments, + const std::vector& oldSegments) { + calls++; + seenNew = newSegments; + ASSERT_EQ(newSegments.size(), oldSegments.size()); + }); + ASSERT_EQ(calls, 1); + ASSERT_TRUE(seenNew.empty()); + + session->close(); + classic->shutdown(); +} From 0c1344805ffb49f6cec4adc0af1f91b38e50e129 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 09:38:06 -0700 Subject: [PATCH 2/5] =?UTF-8?q?st:=20stream=20consumer=20core=20=E2=80=94?= =?UTF-8?q?=20assignment-driven=20Exclusive=20fan-in=20with=20cumulative?= =?UTF-8?q?=20acks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StreamConsumerImpl, the Java v5 ScalableStreamConsumer port. The client enforces no DAG ordering itself — the broker's subscription coordinator withholds child segments from the assignment until every parent drains — so the consumer subscribes Exclusive to exactly the assigned segments (ConsumerAssignmentSession), runs one receive loop per segment into the shared mux ReceiveQueue, and stamps each delivered message with a position vector: a snapshot of every segment's latest-delivered id taken at delivery time, so one cumulative acknowledgment advances all cursors (fanned out as one acknowledgeCumulativeAsync per segment). - Assignment reconcile: released segments close immediately (the shared cursor redelivers unacked messages to the next owner); newly assigned segments retry only rebalance collisions (ConsumerBusy / ConsumerAssignError — the previous Exclusive owner has not released yet) with bounded backoff, and fail fast on anything else. - TopicTerminated closes the segment immediately and drops its bookkeeping; a late cumulative ack for it is a no-op (Java parity — the queue consumer's outstanding-count deferral does not transfer to cumulative acks). - ReceiveQueue::receiveMultiAsync: batch receive collecting until full or deadline (possibly short, including empty on a quiet timeout), with the same executor-hop guard against inline recursion, plus broker-free unit tests. - MessageIdFactory grows the position-vector overload; readCompacted is now accepted for segment topics on the subscribe seam (they are persistent in all but the scheme); AckPolicy::negativeAckRedeliveryDelay is deliberately not wired (a stream consumer has no negative-ack path). - PIP-486 bucket-shared assignments are not supported yet: the subscribe fails loudly when the initial assignment carries bucket ranges. Wired through StreamConsumerCore and StClientImpl::subscribeStreamAsync. --- include/pulsar/st/detail/StreamConsumerCore.h | 2 + lib/ClientImpl.cc | 10 +- lib/st/MessageIdImpl.h | 14 + lib/st/ReceiveQueue.cc | 61 +++ lib/st/ReceiveQueue.h | 16 + lib/st/StClientImpl.cc | 16 +- lib/st/StreamConsumerCore.cc | 58 +++ lib/st/StreamConsumerImpl.cc | 428 ++++++++++++++++++ lib/st/StreamConsumerImpl.h | 131 ++++++ tests/st/StReceiveQueueTest.cc | 94 ++++ 10 files changed, 824 insertions(+), 6 deletions(-) create mode 100644 lib/st/StreamConsumerCore.cc create mode 100644 lib/st/StreamConsumerImpl.cc create mode 100644 lib/st/StreamConsumerImpl.h create mode 100644 tests/st/StReceiveQueueTest.cc diff --git a/include/pulsar/st/detail/StreamConsumerCore.h b/include/pulsar/st/detail/StreamConsumerCore.h index aea16e71..e03599cd 100644 --- a/include/pulsar/st/detail/StreamConsumerCore.h +++ b/include/pulsar/st/detail/StreamConsumerCore.h @@ -34,6 +34,7 @@ namespace pulsar::st { class StreamConsumerImpl; using StreamConsumerImplPtr = std::shared_ptr; 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_; diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc index 043d6d42..a04f95f7 100644 --- a/lib/ClientImpl.cc +++ b/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; diff --git a/lib/st/MessageIdImpl.h b/lib/st/MessageIdImpl.h index 0b3f9daf..31eaf682 100644 --- a/lib/st/MessageIdImpl.h +++ b/lib/st/MessageIdImpl.h @@ -70,6 +70,20 @@ class MessageIdFactory { return MessageId(std::move(impl)); } + /** + * The stream-consumer path: an id that also carries a snapshot of every + * segment's latest-delivered position, so one cumulative ack advances all + * of them. + */ + static MessageId create(const pulsar::MessageId& v4MessageId, std::int64_t segmentId, + std::map positionVector) { + auto impl = std::make_shared(); + impl->v4MessageId = v4MessageId; + impl->segmentId = segmentId; + impl->positionVector = std::move(positionVector); + return MessageId(std::move(impl)); + } + static const std::shared_ptr& impl(const MessageId& id) { return id.impl_; } }; diff --git a/lib/st/ReceiveQueue.cc b/lib/st/ReceiveQueue.cc index b8407a34..2e33f14b 100644 --- a/lib/st/ReceiveQueue.cc +++ b/lib/st/ReceiveQueue.cc @@ -159,6 +159,67 @@ Future ReceiveQueue::offer(MessageImplPtr message) { return capacityPromise.getFuture(); } +Future> ReceiveQueue::receiveMultiAsync(int maxMessages, + std::chrono::milliseconds timeout) { + detail::Promise> promise; + if (maxMessages <= 0) { + promise.setValue({}); + return promise.getFuture(); + } + collectMulti(promise, std::make_shared>(), maxMessages, + std::chrono::steady_clock::now() + timeout); + return promise.getFuture(); +} + +void ReceiveQueue::collectMulti(detail::Promise> promise, + std::shared_ptr> batch, int maxMessages, + std::chrono::steady_clock::time_point deadline) { + std::deque> toSignal; + bool closed = false; + { + std::lock_guard lock(mutex_); + closed = closed_; + while (static_cast(batch->size()) < maxMessages && !buffer_.empty()) { + batch->push_back(std::move(buffer_.front())); + buffer_.pop_front(); + } + if (!closed) toSignal = takeCapacityWaitersIfRoomLocked(); + } + for (auto& waiter : toSignal) waiter.setSuccess(); + + if (closed && batch->empty()) { + promise.setError(Error{ResultAlreadyClosed, "consumer is closed"}); + return; + } + const auto now = std::chrono::steady_clock::now(); + if (closed || static_cast(batch->size()) >= maxMessages || now >= deadline) { + promise.setValue(std::move(*batch)); + return; + } + // Wait for the next message with whatever deadline remains, then collect again. + // The continuation hops through the executor: receiveAsync can complete inline + // when a message races in, and an inline continuation would recurse per message. + auto self = shared_from_this(); + const auto remaining = std::chrono::duration_cast(deadline - now); + receiveAsync(remaining).addListener( + [self, promise, batch, maxMessages, deadline](const Expected& result) { + if (result) { + batch->push_back(*result); + self->executor_->postWork([self, promise, batch, maxMessages, deadline] { + self->collectMulti(promise, batch, maxMessages, deadline); + }); + return; + } + if (result.error().result == ResultTimeout || !batch->empty()) { + // A quiet deadline (or a close racing a partial batch): hand over what + // was collected — possibly nothing. + promise.setValue(std::move(*batch)); + } else { + promise.setError(result.error()); + } + }); +} + void ReceiveQueue::close() { std::map pending; std::deque> waiters; diff --git a/lib/st/ReceiveQueue.h b/lib/st/ReceiveQueue.h index 762ffa0a..54cb62e3 100644 --- a/lib/st/ReceiveQueue.h +++ b/lib/st/ReceiveQueue.h @@ -27,6 +27,7 @@ #include #include #include +#include #include "lib/ExecutorService.h" @@ -51,6 +52,15 @@ class ReceiveQueue : public std::enable_shared_from_this { Future receiveAsync(); Future receiveAsync(std::chrono::milliseconds timeout); + /** + * Receive up to maxMessages messages: wait (up to the deadline) for the first + * one, then opportunistically drain whatever is already buffered, repeating + * until the batch is full or the deadline elapses. May complete with fewer + * than maxMessages — including zero on a quiet timeout. + */ + Future> receiveMultiAsync(int maxMessages, + std::chrono::milliseconds timeout); + /** Deliver a message; the returned future completes when there is room for the next offer. */ Future offer(MessageImplPtr message); @@ -62,6 +72,12 @@ class ReceiveQueue : public std::enable_shared_from_this { // the returned promises must be completed after releasing it. std::deque> takeCapacityWaitersIfRoomLocked(); + // One receiveMultiAsync collection round: greedily drain what is buffered, then + // wait for the next message with the remaining deadline and go again. + void collectMulti(detail::Promise> promise, + std::shared_ptr> batch, int maxMessages, + std::chrono::steady_clock::time_point deadline); + // A parked receive: the promise to complete and — for timed receives — the timeout timer, // cancelled when a message (or close) wins the race so idle timers don't accumulate. struct PendingReceive { diff --git a/lib/st/StClientImpl.cc b/lib/st/StClientImpl.cc index 1c6f4467..256d26b9 100644 --- a/lib/st/StClientImpl.cc +++ b/lib/st/StClientImpl.cc @@ -23,6 +23,7 @@ #include "QueueConsumerImpl.h" #include "StProducerImpl.h" +#include "StreamConsumerImpl.h" namespace pulsar::st { @@ -62,9 +63,18 @@ Future ClientImpl::createProducerAsync(ProducerConfig conf // value (the sink the real implementation will move from), but as a stub it does not // consume the config yet — hence the value-param suppressions. -// NOLINTNEXTLINE(performance-unnecessary-value-param) -Future ClientImpl::subscribeStreamAsync(StreamConsumerConfig) { - return notImplementedYet("subscribeStream"); +Future ClientImpl::subscribeStreamAsync(StreamConsumerConfig config) { + auto impl = std::make_shared(classic_, std::move(config)); + detail::Promise promise; + // Keep the impl alive until start() resolves; on success mint the public core over it. + impl->start().addListener([impl, promise](const Expected& result) { + if (result) { + promise.setValue(detail::StreamConsumerCore{impl}); + } else { + promise.setError(result.error()); + } + }); + return promise.getFuture(); } Future ClientImpl::subscribeQueueAsync(QueueConsumerConfig config) { diff --git a/lib/st/StreamConsumerCore.cc b/lib/st/StreamConsumerCore.cc new file mode 100644 index 00000000..37b82b19 --- /dev/null +++ b/lib/st/StreamConsumerCore.cc @@ -0,0 +1,58 @@ +/** + * 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 +#include + +#include "StreamConsumerImpl.h" + +namespace pulsar::st::detail { + +// Thin forwarders to the hidden StreamConsumerImpl. The receive paths map the impl's +// MessageImplPtr to a MessageCore — the mapping lambdas run in this member context, +// which is a friend of MessageCore, so they can reach MessageCore's private constructor. +Future StreamConsumerCore::receiveAsync() const { + return impl_->receiveAsync().thenApply( + [](const MessageImplPtr& message) { return MessageCore{message}; }); +} +Future StreamConsumerCore::receiveAsync(std::chrono::milliseconds timeout) const { + return impl_->receiveAsync(timeout).thenApply( + [](const MessageImplPtr& message) { return MessageCore{message}; }); +} +Future> StreamConsumerCore::receiveMultiAsync( + int maxMessages, std::chrono::milliseconds timeout) const { + return impl_->receiveMultiAsync(maxMessages, timeout) + .thenApply([](const std::vector& messages) { + std::vector cores; + cores.reserve(messages.size()); + for (const auto& message : messages) cores.push_back(MessageCore{message}); + return cores; + }); +} +void StreamConsumerCore::acknowledgeCumulative(const MessageId& id) const { + impl_->acknowledgeCumulative(id); +} +void StreamConsumerCore::acknowledgeCumulative(const MessageId& id, const Transaction& txn) const { + impl_->acknowledgeCumulative(id, txn); +} +Future StreamConsumerCore::closeAsync() const { return impl_->closeAsync(); } +std::string_view StreamConsumerCore::topic() const { return impl_->topic(); } +std::string_view StreamConsumerCore::subscription() const { return impl_->subscription(); } +std::string_view StreamConsumerCore::consumerName() const { return impl_->consumerName(); } + +} // namespace pulsar::st::detail diff --git a/lib/st/StreamConsumerImpl.cc b/lib/st/StreamConsumerImpl.cc new file mode 100644 index 00000000..5ad6ee9d --- /dev/null +++ b/lib/st/StreamConsumerImpl.cc @@ -0,0 +1,428 @@ +/** + * 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 "StreamConsumerImpl.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "MessageIdImpl.h" +#include "MessageImpl.h" +#include "lib/LogUtils.h" + +DECLARE_LOG_OBJECT() + +namespace pulsar::st { + +namespace { + +pulsar::InitialPosition toClassicInitialPosition(SubscriptionInitialPosition position) { + return position == SubscriptionInitialPosition::Earliest ? pulsar::InitialPositionEarliest + : pulsar::InitialPositionLatest; +} + +// Close a segment consumer once its creation future resolves (a no-op if creation failed). +void closeWhenReady(Future future) { + future.addListener([](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.closeAsync([](pulsar::Result) {}); + } + }); +} + +// The consumer name is the controller's registration key, so one is always needed: +// default to "v5-stream-" + 8 random hex chars when unset (Java parity). +std::string defaultConsumerName() { + static const char kHex[] = "0123456789abcdef"; + std::random_device rd; + std::string suffix(8, '0'); + for (auto& c : suffix) { + c = kHex[rd() % 16]; + } + return "v5-stream-" + suffix; +} + +// Rebalance handoffs collide briefly by design: the previous Exclusive owner has not +// released the segment yet (ConsumerBusy), or sticky ranges still overlap +// (ConsumerAssignError). Only these are worth retrying; anything else is a real +// error that retrying would only hide. +bool isRebalanceCollision(Result result) { + return result == ResultConsumerBusy || result == ResultConsumerAssignError; +} + +} // namespace + +StreamConsumerImpl::StreamConsumerImpl(pulsar::ClientImplPtr classic, StreamConsumerConfig config) + : classic_(std::move(classic)), + config_(std::move(config)), + topic_(config_.topic), + subscription_(config_.subscriptionName), + consumerName_(config_.consumerName.value_or(defaultConsumerName())), + executor_(classic_->getIOExecutorProvider()->get()), + receiveQueue_(std::make_shared(executor_, kReceiveQueueCapacity)) {} + +Future StreamConsumerImpl::start() { + if (config_.useNamespace) { + startPromise_.setError(Error{ResultOperationNotSupported, + "namespace-mode subscriptions are not implemented yet in the " + "scalable-topics client"}); + return startPromise_.getFuture(); + } + session_ = std::make_shared( + classic_, config_.topic, config_.subscriptionName, consumerName_, + pulsar::ScalableConsumerType_STREAM); + std::weak_ptr weak = weak_from_this(); + session_->setListener([weak](const std::vector& newSegments, + const std::vector& oldSegments) { + if (auto self = weak.lock()) self->onAssignmentChange(newSegments, oldSegments); + }); + // The listener drives the success path (the session applies the initial assignment + // through it before this future resolves); here only a failure is surfaced. + session_->start().addListener([weak](const Expected>& result) { + if (!result) { + if (auto self = weak.lock()) self->startPromise_.setError(result.error()); + } + }); + return startPromise_.getFuture(); +} + +void StreamConsumerImpl::onAssignmentChange(const std::vector& newSegments, + const std::vector& /*oldSegments*/) { + std::vector> retired; + std::vector toAdd; + std::vector bucketShared; + bool first = false; + { + std::lock_guard lock(mutex_); + first = !sawFirstAssignment_; + sawFirstAssignment_ = true; + currentAssignment_ = newSegments; + + std::unordered_set targetIds; + for (const auto& assigned : newSegments) targetIds.insert(assigned.segment.segmentId); + for (auto it = segmentConsumers_.begin(); it != segmentConsumers_.end();) { + if (targetIds.find(it->first) == targetIds.end()) { + // Released by a rebalance: close immediately. Unacked in-flight messages + // are redelivered to the segment's next owner from the shared cursor. + retired.push_back(std::move(it->second)); + latestDelivered_.erase(static_cast(it->first)); + it = segmentConsumers_.erase(it); + } else { + ++it; + } + } + for (const auto& assigned : newSegments) { + if (segmentConsumers_.find(assigned.segment.segmentId) != segmentConsumers_.end()) { + continue; + } + if (!assigned.ownedBucketRanges.empty()) { + bucketShared.push_back(assigned.segment.segmentId); + } else { + toAdd.push_back(assigned); + } + } + } + + for (auto& future : retired) closeWhenReady(future); + + if (!bucketShared.empty()) { + // PIP-486 bucket-sharing (consumers outnumbering segments) is not supported yet. + if (first) { + startPromise_.setError( + Error{ResultOperationNotSupported, + "bucket-shared segment assignments (PIP-486) are not supported yet in the " + "scalable-topics client; use at most one stream consumer per segment"}); + return; + } + for (auto segmentId : bucketShared) { + LOG_ERROR("[" << topic_ << "] segment " << segmentId + << " was assigned bucket-shared (PIP-486), which is not supported yet; " + "skipping it"); + } + } + + if (first) { + if (toAdd.empty()) { + startPromise_.setSuccess(); + return; + } + auto remaining = std::make_shared>(static_cast(toAdd.size())); + for (const auto& assigned : toAdd) { + getOrCreateSegmentConsumerAsync(assigned).addListener( + [self = shared_from_this(), remaining](const Expected& result) { + if (!result) { + self->startPromise_.setError(result.error()); // first error wins (idempotent) + return; + } + if (remaining->fetch_sub(1) == 1) self->startPromise_.setSuccess(); + }); + } + } else { + for (const auto& assigned : toAdd) subscribeSegmentWithRetry(assigned, /*attempt*/ 0); + } +} + +pulsar::ConsumerConfiguration StreamConsumerImpl::buildSegmentConfiguration( + const AssignedSegment& assigned) const { + // Build a FRESH config every time (pulsar::ConsumerConfiguration's copy ctor shares its impl). + pulsar::ConsumerConfiguration conf; + conf.setConsumerType(pulsar::ConsumerExclusive); + conf.setSchema(config_.schema); + conf.setSubscriptionInitialPosition(toClassicInitialPosition(config_.initialPosition)); + conf.setConsumerName(consumerName_ + "-seg-" + std::to_string(assigned.segment.segmentId)); + if (config_.ackPolicy.groupTime) { + conf.setAckGroupingTimeMs(static_cast(config_.ackPolicy.groupTime->count())); + } + // AckPolicy::negativeAckRedeliveryDelay is deliberately not wired: a stream + // consumer has no negative-ack path (documented on the config field). + if (config_.readCompacted) { + conf.setReadCompacted(*config_.readCompacted); + } + if (config_.replicateSubscriptionState) { + conf.setReplicateSubscriptionStateEnabled(*config_.replicateSubscriptionState); + } + if (!config_.subscriptionProperties.empty()) { + conf.setSubscriptionProperties(config_.subscriptionProperties); + } + for (const auto& [key, value] : config_.properties) conf.setProperty(key, value); + if (assigned.segment.isLegacy()) conf.setProperty("__pulsar.v5.managed", "true"); + return conf; +} + +Future StreamConsumerImpl::getOrCreateSegmentConsumerAsync( + const AssignedSegment& assigned) { + detail::Promise promise; + const std::uint64_t segmentId = assigned.segment.segmentId; + { + std::lock_guard lock(mutex_); + if (auto it = segmentConsumers_.find(segmentId); it != segmentConsumers_.end()) { + return it->second; + } + segmentConsumers_.insert_or_assign(segmentId, promise.getFuture()); + } + + const pulsar::ConsumerConfiguration conf = buildSegmentConfiguration(assigned); + const std::string attachTopic = assigned.segment.attachTopicName(); + auto self = shared_from_this(); + classic_->subscribeSegmentAsync( + attachTopic, config_.subscriptionName, conf, + [self, promise, segmentId](std::variant result) { + if (auto* consumer = std::get_if(&result)) { + // pulsar::Consumer is a copyable handle (its virtual dtor suppresses the + // move ctor), so this is a shared-impl copy, not a deep copy. + pulsar::Consumer c = *consumer; + self->startReceiveLoop(c, segmentId); + promise.setValue(c); + } else { + // Evict the failed subscribe so a later attempt (retry or assignment + // push) can re-create it. + { + std::lock_guard lock(self->mutex_); + self->segmentConsumers_.erase(segmentId); + } + promise.setError(std::get(result)); + } + }); + return promise.getFuture(); +} + +bool StreamConsumerImpl::isSegmentStillAssignedLocked(std::uint64_t segmentId) const { + for (const auto& assigned : currentAssignment_) { + if (assigned.segment.segmentId == segmentId) return true; + } + return false; +} + +void StreamConsumerImpl::subscribeSegmentWithRetry(const AssignedSegment& assigned, int attempt) { + std::weak_ptr weak = weak_from_this(); + getOrCreateSegmentConsumerAsync(assigned).addListener( + [weak, assigned, attempt](const Expected& result) { + auto self = weak.lock(); + if (result || !self || self->closed_.load()) return; + if (!isRebalanceCollision(result.error().result)) { + LOG_ERROR("[" << self->topic_ << "] segment " << assigned.segment.segmentId + << " subscribe failed (" << result.error() + << "); not a rebalance collision, waiting for the next assignment"); + return; + } + if (attempt + 1 >= kSubscribeRetryMaxAttempts) { + LOG_ERROR("[" << self->topic_ << "] segment " << assigned.segment.segmentId + << " subscribe still colliding after " << kSubscribeRetryMaxAttempts + << " attempts; giving up until the next assignment: " << result.error()); + return; + } + { + std::lock_guard lock(self->mutex_); + if (!self->isSegmentStillAssignedLocked(assigned.segment.segmentId)) return; + } + LOG_INFO("[" << self->topic_ << "] segment " << assigned.segment.segmentId + << " is still held by its previous owner; retrying, attempt " << (attempt + 1) + << " of " << kSubscribeRetryMaxAttempts); + auto timer = self->executor_->createDeadlineTimer(); + const std::int64_t delayMs = + std::min(100 * (attempt + 1), kSubscribeRetryMaxBackoffMs); + timer->expires_from_now(std::chrono::milliseconds(delayMs)); + // Weak ref: closeAsync() does not cancel these timers (`timer` keeps itself alive). + timer->async_wait([weak, assigned, attempt, timer](const ASIO_ERROR& ec) { + auto self = weak.lock(); + if (ec || !self || self->closed_.load()) return; + self->subscribeSegmentWithRetry(assigned, attempt + 1); + }); + }); +} + +void StreamConsumerImpl::startReceiveLoop(pulsar::Consumer consumer, std::uint64_t segmentId) { + if (closed_.load()) return; + auto self = shared_from_this(); + consumer.receiveAsync([self, consumer, segmentId](pulsar::Result result, const pulsar::Message& message) { + if (result != pulsar::ResultOk) { + if (result == pulsar::ResultTopicTerminated) { + // The sealed segment is fully drained: close immediately and drop its + // bookkeeping. A late cumulative ack carrying this segment's position is + // a no-op — the cursor is already at the end. (The queue consumer's + // deferred close does not transfer here: one cumulative ack settles an + // unbounded prefix, so there is no per-message outstanding count.) + { + std::lock_guard lock(self->mutex_); + self->segmentConsumers_.erase(segmentId); + self->latestDelivered_.erase(static_cast(segmentId)); + } + pulsar::Consumer done = consumer; + done.closeAsync([](pulsar::Result) {}); + } + // Otherwise (AlreadyClosed / consumer closing) just stop the loop. + return; + } + // Snapshot the position vector AT DELIVERY TIME, inside the segment loop: every + // delivered message carries where all segments stood when it was handed over, so + // acknowledging it cumulatively advances exactly what had been delivered by then. + std::map positionVector; + { + std::lock_guard lock(self->mutex_); + self->latestDelivered_[static_cast(segmentId)] = message.getMessageId(); + positionVector = self->latestDelivered_; + } + MessageId id = MessageIdFactory::create(message.getMessageId(), static_cast(segmentId), + std::move(positionVector)); + // Report the scalable topic as the source, not the internal segment:// backing topic. + auto messageImpl = std::make_shared(message, std::move(id), self->topic_); + // Re-arm only once the fan-in queue has room, and hop through the executor so + // the per-message chain is a loop rather than recursion (see QueueConsumerImpl). + self->receiveQueue_->offer(std::move(messageImpl)) + .addListener([self, consumer, segmentId](const Expected&) { + self->executor_->postWork( + [self, consumer, segmentId] { self->startReceiveLoop(consumer, segmentId); }); + }); + }); +} + +Future StreamConsumerImpl::receiveAsync() { return receiveQueue_->receiveAsync(); } + +Future StreamConsumerImpl::receiveAsync(std::chrono::milliseconds timeout) { + return receiveQueue_->receiveAsync(timeout); +} + +Future> StreamConsumerImpl::receiveMultiAsync( + int maxMessages, std::chrono::milliseconds timeout) { + return receiveQueue_->receiveMultiAsync(maxMessages, timeout); +} + +void StreamConsumerImpl::acknowledgeCumulative(const MessageId& id) { + const auto& impl = MessageIdFactory::impl(id); + if (!impl) return; + // Fan the position vector out as one cumulative ack per segment. This may also + // advance segments past messages still sitting unread in the mux queue — the + // vector records what had been DELIVERED into the queue when this message was, + // not what the application has read; that is the contract, not a defect. + auto positions = impl->positionVector; + if (positions.empty() && impl->segmentId != MessageIdImpl::kNoSegment) { + // An id without a vector (not minted by this consumer): ack its own segment. + positions.emplace(impl->segmentId, impl->v4MessageId); + } + for (const auto& [segmentId, position] : positions) { + std::optional> future; + { + std::lock_guard lock(mutex_); + auto it = segmentConsumers_.find(static_cast(segmentId)); + if (it != segmentConsumers_.end()) future = it->second; + } + if (!future) continue; // drained or released: the cursor no longer needs this ack + const pulsar::MessageId v4 = position; + future->addListener([v4](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.acknowledgeCumulativeAsync(v4, [](pulsar::Result) {}); + } + }); + } +} + +void StreamConsumerImpl::acknowledgeCumulative(const MessageId& /*id*/, const Transaction& /*txn*/) { + // Transactions are not implemented yet in the scalable-topics client, and an ack is + // fire-and-forget void (no error channel). Drop it — the cursor simply does not advance. + LOG_WARN("[" << topic_ << "] transactional acknowledge is not implemented yet; dropping the ack"); +} + +Future StreamConsumerImpl::closeAsync() { + if (closed_.exchange(true)) { + detail::Promise promise; + promise.setSuccess(); // idempotent + return promise.getFuture(); + } + if (session_) session_->close(); + if (receiveQueue_) receiveQueue_->close(); // fail pending receives + + std::vector> consumers; + { + std::lock_guard lock(mutex_); + consumers.reserve(segmentConsumers_.size()); + for (auto& [segmentId, future] : segmentConsumers_) consumers.push_back(future); + segmentConsumers_.clear(); + latestDelivered_.clear(); + currentAssignment_.clear(); + } + + detail::Promise promise; + auto remaining = std::make_shared>(static_cast(consumers.size()) + 1); + auto finishOne = [promise, remaining]() { + if (remaining->fetch_sub(1) == 1) promise.setSuccess(); + }; + for (auto& future : consumers) { + future.addListener([finishOne](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.closeAsync([finishOne](pulsar::Result) { finishOne(); }); // swallow errors + } else { + finishOne(); + } + }); + } + finishOne(); + return promise.getFuture(); +} + +} // namespace pulsar::st diff --git a/lib/st/StreamConsumerImpl.h b/lib/st/StreamConsumerImpl.h new file mode 100644 index 00000000..a6d4a12b --- /dev/null +++ b/lib/st/StreamConsumerImpl.h @@ -0,0 +1,131 @@ +/** + * 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 +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "ConsumerAssignmentSession.h" +#include "ReceiveQueue.h" +#include "lib/ClientImpl.h" +#include "lib/ExecutorService.h" + +namespace pulsar::st { + +/** + * The scalable-topics stream consumer (single scalable topic): ordered consumption + * over the segment DAG, a port of the Java v5 ScalableStreamConsumer. + * + * Unlike the queue consumer, the client enforces no ordering itself: the broker's + * subscription coordinator withholds a child segment from the assignment until every + * parent has drained (per-subscription backlog reaching zero), so per-key order + * across splits and merges holds as long as the application keeps acknowledging. + * The client's job is to subscribe — Exclusive — to exactly the assigned segments + * (delivered by the ConsumerAssignmentSession, updated on every rebalance), run one + * receive loop per segment fanning into the shared mux ReceiveQueue, and stamp every + * delivered message with a snapshot of all segments' latest-delivered positions (the + * position vector), so one cumulative acknowledgment advances every segment's cursor. + * + * A segment reporting TopicTerminated (sealed and fully drained) is closed and its + * bookkeeping dropped immediately; a late cumulative ack for it is a no-op — the + * cursor is already at the end (Java parity; the queue consumer's outstanding-count + * deferral does not transfer, because one cumulative ack settles an unbounded + * prefix). + * + * PIP-486 bucket-shared segments (non-empty ownedBucketRanges) are not supported + * yet: the whole-segment Exclusive path covers every assignment the controller + * produces while consumers do not outnumber segments. + */ +class StreamConsumerImpl : public std::enable_shared_from_this { + public: + StreamConsumerImpl(pulsar::ClientImplPtr classic, StreamConsumerConfig config); + + /** Register with the controller and subscribe the initially assigned segments. */ + Future start(); + + Future receiveAsync(); + Future receiveAsync(std::chrono::milliseconds timeout); + Future> receiveMultiAsync(int maxMessages, + std::chrono::milliseconds timeout); + void acknowledgeCumulative(const MessageId& id); + void acknowledgeCumulative(const MessageId& id, const Transaction& txn); + Future closeAsync(); + + std::string_view topic() const { return topic_; } + std::string_view subscription() const { return subscription_; } + std::string_view consumerName() const { return consumerName_; } + + private: + // How many messages the fan-in queue buffers before back-pressuring the segment receive loops. + static constexpr std::size_t kReceiveQueueCapacity = 1000; + // Rebalance handoffs are expected to collide briefly (the previous Exclusive owner + // has not released the segment yet): retry those within a bounded backoff and fail + // everything else fast. Constants mirror the producer's send retry. + static constexpr int kSubscribeRetryMaxAttempts = 10; + static constexpr std::int64_t kSubscribeRetryMaxBackoffMs = 500; + + pulsar::ConsumerConfiguration buildSegmentConfiguration(const AssignedSegment& assigned) const; + Future getOrCreateSegmentConsumerAsync(const AssignedSegment& assigned); + // getOrCreateSegmentConsumerAsync plus a bounded backoff retry on the rebalance + // collisions (ConsumerBusy / ConsumerAssignError), used off the start path. + void subscribeSegmentWithRetry(const AssignedSegment& assigned, int attempt); + void startReceiveLoop(pulsar::Consumer consumer, std::uint64_t segmentId); + + void onAssignmentChange(const std::vector& newSegments, + const std::vector& oldSegments); + // Whether the segment is still in the current assignment. Caller holds mutex_. + bool isSegmentStillAssignedLocked(std::uint64_t segmentId) const; + + pulsar::ClientImplPtr classic_; + const StreamConsumerConfig config_; + const std::string topic_; + const std::string subscription_; + // The controller registration key: defaults to "v5-stream-" + 8 random hex chars + // when the application did not set one (Java parity). + const std::string consumerName_; + const pulsar::ExecutorServicePtr executor_; + ConsumerAssignmentSessionPtr session_; + ReceiveQueuePtr receiveQueue_; + detail::Promise startPromise_; + std::atomic closed_{false}; + + mutable std::mutex mutex_; + bool sawFirstAssignment_ = false; // guarded by mutex_ + std::vector currentAssignment_; // guarded by mutex_ + std::unordered_map> segmentConsumers_; // guarded by mutex_ + // Every segment's latest-delivered position, snapshotted into each delivered + // message's position vector at delivery time. Guarded by mutex_. + std::map latestDelivered_; +}; + +using StreamConsumerImplPtr = std::shared_ptr; + +} // namespace pulsar::st diff --git a/tests/st/StReceiveQueueTest.cc b/tests/st/StReceiveQueueTest.cc new file mode 100644 index 00000000..bf407587 --- /dev/null +++ b/tests/st/StReceiveQueueTest.cc @@ -0,0 +1,94 @@ +/** + * 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 +#include + +#include +#include + +#include "lib/ExecutorService.h" +#include "lib/st/MessageIdImpl.h" +#include "lib/st/MessageImpl.h" +#include "lib/st/ReceiveQueue.h" + +// Broker-free tests for the fan-in mux queue's batch receive: greedy drain of what +// is buffered, deadline behavior, and close. The single-message paths are covered +// end-to-end by the queue-consumer tests. + +using namespace pulsar::st; + +namespace { + +MessageImplPtr makeMessage(int i) { + auto classic = pulsar::MessageBuilder().setContent("m-" + std::to_string(i)).build(); + return std::make_shared(classic, MessageIdFactory::create(pulsar::MessageId::earliest(), 0)); +} + +pulsar::ExecutorServicePtr makeExecutor() { + static auto provider = std::make_shared(1); + return provider->get(); +} + +} // namespace + +TEST(StReceiveQueueTest, testReceiveMultiDrainsBufferedWithoutWaiting) { + auto queue = std::make_shared(makeExecutor(), 100); + for (int i = 0; i < 5; i++) { + queue->offer(makeMessage(i)); + } + auto batch = queue->receiveMultiAsync(3, std::chrono::seconds(10)).get(); + ASSERT_TRUE(batch); + ASSERT_EQ(batch->size(), 3u); + + auto rest = queue->receiveMultiAsync(10, std::chrono::milliseconds(50)).get(); + ASSERT_TRUE(rest); + // Fewer than asked: the deadline elapsed after draining what was buffered. + ASSERT_EQ(rest->size(), 2u); + queue->close(); +} + +TEST(StReceiveQueueTest, testReceiveMultiTimesOutEmpty) { + auto queue = std::make_shared(makeExecutor(), 100); + auto batch = queue->receiveMultiAsync(4, std::chrono::milliseconds(50)).get(); + ASSERT_TRUE(batch); + ASSERT_TRUE(batch->empty()) << "a quiet deadline yields an empty batch, not an error"; + queue->close(); +} + +TEST(StReceiveQueueTest, testReceiveMultiWaitsForFirstThenDrains) { + auto queue = std::make_shared(makeExecutor(), 100); + // The batch collects until full or deadline; with two messages arriving after the + // receive parked, the short deadline hands back a partial batch of both. + auto future = queue->receiveMultiAsync(5, std::chrono::milliseconds(500)); + queue->offer(makeMessage(0)); + queue->offer(makeMessage(1)); + auto batch = future.get(); + ASSERT_TRUE(batch); + ASSERT_GE(batch->size(), 1u); + ASSERT_LE(batch->size(), 2u); + queue->close(); +} + +TEST(StReceiveQueueTest, testReceiveMultiFailsWhenClosedEmpty) { + auto queue = std::make_shared(makeExecutor(), 100); + queue->close(); + auto batch = queue->receiveMultiAsync(4, std::chrono::seconds(1)).get(); + ASSERT_FALSE(batch); + ASSERT_EQ(batch.error().result, pulsar::ResultAlreadyClosed); +} From 749b037b9b4ef902507b8ab79aed4791e1bf9d96 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 09:42:32 -0700 Subject: [PATCH 3/5] st: stream consumer e2e + shared admin harness for the e2e suites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three end-to-end tests against a real scalable-topics broker: - testOrderedRoundTripAndCumulativeAck: one segment, one Exclusive consumer — delivery is exactly the publish order, and one cumulative ack of the last message settles the whole stream (a reattached consumer receives nothing). - testDrainsSealedParentBeforeChildren: the DAG-replay scenario. Everything produced before a split sits in the sealed parent; the broker withholds the children from the assignment until the parent is drained for this subscription, so acking as we go is what unblocks them — and every parent message must arrive before any child message. Exercises the controller registration, the initial assignment, the drain detection, and the pushed assignment update end to end. - testCumulativeAckCoversAllSegments: two initial segments drained through receiveMulti without intermediate acks; acknowledging only the final message advances both cursors through its position vector. The admin REST helpers the three e2e suites had each carried are extracted into tests/st/StE2EAdmin.h (the dedup flagged in #605). --- tests/st/StE2EAdmin.h | 88 +++++++++ tests/st/StProducerE2ETest.cc | 57 +----- tests/st/StQueueConsumerE2ETest.cc | 50 +---- tests/st/StStreamConsumerE2ETest.cc | 276 ++++++++++++++++++++++++++++ 4 files changed, 368 insertions(+), 103 deletions(-) create mode 100644 tests/st/StE2EAdmin.h create mode 100644 tests/st/StStreamConsumerE2ETest.cc diff --git a/tests/st/StE2EAdmin.h b/tests/st/StE2EAdmin.h new file mode 100644 index 00000000..118676f3 --- /dev/null +++ b/tests/st/StE2EAdmin.h @@ -0,0 +1,88 @@ +/** + * 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 + +// Shared harness helpers for the scalable-topics end-to-end tests: the PULSAR_ST_E2E +// gate, broker endpoints, fresh per-run topic names, and the admin REST calls that +// create topics and drive split/merge at the exact point a test needs them. + +#include +#include +#include +#include +#include +#include + +#include "tests/HttpHelper.h" + +namespace st_e2e { + +inline bool e2eEnabled() { return std::getenv("PULSAR_ST_E2E") != nullptr; } + +inline std::string serviceUrl() { + const char* url = std::getenv("PULSAR_ST_E2E_SERVICE_URL"); + return url != nullptr ? url : "pulsar://localhost:6650"; +} + +inline std::string adminUrl() { + const char* url = std::getenv("PULSAR_ST_E2E_ADMIN_URL"); + return url != nullptr ? url : "http://localhost:8080"; +} + +// A fresh topic name per test run so tests never collide with one another or with a +// topic left on a reused broker (the same convention the classic tests use). +inline std::string uniqueName(const std::string& prefix) { + static int counter = 0; + return prefix + "-" + std::to_string(std::time(nullptr)) + "-" + std::to_string(counter++); +} + +inline std::string topicUrl(const std::string& name) { return "topic://public/default/" + name; } + +// The admin REST base for a scalable topic under public/default. +inline std::string scalablePath(const std::string& name) { + return adminUrl() + "/admin/v2/scalable/public/default/" + name; +} + +// Create a scalable topic with the given number of initial segments. Retries while +// the scalable-topics controller finishes coming up after broker start (only the +// first test waits). +inline bool createScalableTopic(const std::string& name, int numInitialSegments = 1) { + const std::string url = scalablePath(name) + "?numInitialSegments=" + std::to_string(numInitialSegments); + for (int attempt = 0; attempt < 30; attempt++) { + const int code = makePutRequest(url, ""); + if (code >= 200 && code < 300) return true; + std::this_thread::sleep_for(std::chrono::seconds(1)); + } + return false; +} + +// Split a segment into two half-range children (POST .../split/{segmentId}). +inline bool splitSegment(const std::string& name, std::int64_t segmentId) { + const int code = makePostRequest(scalablePath(name) + "/split/" + std::to_string(segmentId), ""); + return code >= 200 && code < 300; +} + +// Merge two segments back into one full-range child (POST .../merge/{segmentId1}/{segmentId2}). +inline bool mergeSegments(const std::string& name, std::int64_t segmentId1, std::int64_t segmentId2) { + const int code = makePostRequest( + scalablePath(name) + "/merge/" + std::to_string(segmentId1) + "/" + std::to_string(segmentId2), ""); + return code >= 200 && code < 300; +} + +} // namespace st_e2e diff --git a/tests/st/StProducerE2ETest.cc b/tests/st/StProducerE2ETest.cc index f83f93bc..8f4e346c 100644 --- a/tests/st/StProducerE2ETest.cc +++ b/tests/st/StProducerE2ETest.cc @@ -27,71 +27,18 @@ #include #include -#include -#include #include #include -#include #include #include "lib/st/MessageIdImpl.h" -#include "tests/HttpHelper.h" +#include "tests/st/StE2EAdmin.h" using namespace pulsar::st; +using namespace st_e2e; namespace { -bool e2eEnabled() { return std::getenv("PULSAR_ST_E2E") != nullptr; } - -std::string serviceUrl() { - const char* url = std::getenv("PULSAR_ST_E2E_SERVICE_URL"); - return url != nullptr ? url : "pulsar://localhost:6650"; -} - -std::string adminUrl() { - const char* url = std::getenv("PULSAR_ST_E2E_ADMIN_URL"); - return url != nullptr ? url : "http://localhost:8080"; -} - -// A fresh topic name per test run so tests never collide with one another or with a topic left on a -// reused broker (the same convention the classic tests use). -std::string uniqueName(const std::string& prefix) { - static int counter = 0; - return prefix + "-" + std::to_string(std::time(nullptr)) + "-" + std::to_string(counter++); -} - -std::string topicUrl(const std::string& name) { return "topic://public/default/" + name; } - -// The admin REST base for a scalable topic under public/default. -std::string scalablePath(const std::string& name) { - return adminUrl() + "/admin/v2/scalable/public/default/" + name; -} - -// Create a scalable topic with the given number of initial segments. Retries while the -// scalable-topics controller finishes coming up after broker start (only the first test waits). -bool createScalableTopic(const std::string& name, int numInitialSegments = 1) { - const std::string url = scalablePath(name) + "?numInitialSegments=" + std::to_string(numInitialSegments); - for (int attempt = 0; attempt < 30; attempt++) { - const int code = makePutRequest(url, ""); - if (code >= 200 && code < 300) return true; - std::this_thread::sleep_for(std::chrono::seconds(1)); - } - return false; -} - -// Split a segment into two half-range children (POST .../split/{segmentId}). -bool splitSegment(const std::string& name, std::int64_t segmentId) { - const int code = makePostRequest(scalablePath(name) + "/split/" + std::to_string(segmentId), ""); - return code >= 200 && code < 300; -} - -// Merge two segments back into one full-range child (POST .../merge/{segmentId1}/{segmentId2}). -bool mergeSegments(const std::string& name, std::int64_t segmentId1, std::int64_t segmentId2) { - const int code = makePostRequest( - scalablePath(name) + "/merge/" + std::to_string(segmentId1) + "/" + std::to_string(segmentId2), ""); - return code >= 200 && code < 300; -} - // The segment id carried by a produced message id (the whole point of the mapping). std::int64_t segmentIdOf(const MessageId& id) { const auto& impl = MessageIdFactory::impl(id); diff --git a/tests/st/StQueueConsumerE2ETest.cc b/tests/st/StQueueConsumerE2ETest.cc index eef024fc..9636ef24 100644 --- a/tests/st/StQueueConsumerE2ETest.cc +++ b/tests/st/StQueueConsumerE2ETest.cc @@ -25,64 +25,18 @@ #include #include -#include -#include #include #include -#include #include #include "lib/st/MessageIdImpl.h" -#include "tests/HttpHelper.h" +#include "tests/st/StE2EAdmin.h" using namespace pulsar::st; +using namespace st_e2e; namespace { -bool e2eEnabled() { return std::getenv("PULSAR_ST_E2E") != nullptr; } - -std::string serviceUrl() { - const char* url = std::getenv("PULSAR_ST_E2E_SERVICE_URL"); - return url != nullptr ? url : "pulsar://localhost:6650"; -} - -std::string adminUrl() { - const char* url = std::getenv("PULSAR_ST_E2E_ADMIN_URL"); - return url != nullptr ? url : "http://localhost:8080"; -} - -// A fresh topic name per test run so tests never collide with one another or with a topic left on a -// reused broker (the same convention the classic tests use). -std::string uniqueName(const std::string& prefix) { - static int counter = 0; - return prefix + "-" + std::to_string(std::time(nullptr)) + "-" + std::to_string(counter++); -} - -std::string topicUrl(const std::string& name) { return "topic://public/default/" + name; } - -// The admin REST base for a scalable topic under public/default. -std::string scalablePath(const std::string& name) { - return adminUrl() + "/admin/v2/scalable/public/default/" + name; -} - -// Create a scalable topic with the given number of initial segments. Retries while the -// scalable-topics controller finishes coming up after broker start (only the first test waits). -bool createScalableTopic(const std::string& name, int numInitialSegments = 1) { - const std::string url = scalablePath(name) + "?numInitialSegments=" + std::to_string(numInitialSegments); - for (int attempt = 0; attempt < 30; attempt++) { - const int code = makePutRequest(url, ""); - if (code >= 200 && code < 300) return true; - std::this_thread::sleep_for(std::chrono::seconds(1)); - } - return false; -} - -// Split a segment into two half-range children (POST .../split/{segmentId}). -bool splitSegment(const std::string& name, std::int64_t segmentId) { - const int code = makePostRequest(scalablePath(name) + "/split/" + std::to_string(segmentId), ""); - return code >= 200 && code < 300; -} - // The segment id carried by a received message id — the fan-in stamps it on every message. std::int64_t segmentIdOf(const MessageId& id) { const auto& impl = MessageIdFactory::impl(id); diff --git a/tests/st/StStreamConsumerE2ETest.cc b/tests/st/StStreamConsumerE2ETest.cc new file mode 100644 index 00000000..fbd128cf --- /dev/null +++ b/tests/st/StStreamConsumerE2ETest.cc @@ -0,0 +1,276 @@ +/** + * 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. + */ +// End-to-end stream-consumer tests against a real scalable-topics broker: ordered +// consumption over the segment DAG through the controller-assigned Exclusive +// per-segment consumers, and the position-vector cumulative acknowledgment. Gated on +// PULSAR_ST_E2E like the other scalable e2e suites. +#include +#include + +#include +#include +#include +#include +#include + +#include "lib/st/MessageIdImpl.h" +#include "tests/st/StE2EAdmin.h" + +using namespace pulsar::st; +using namespace st_e2e; + +namespace { + +// The segment id carried by a received message id. +std::int64_t segmentIdOf(const MessageId& id) { + const auto& impl = MessageIdFactory::impl(id); + return impl ? impl->segmentId : MessageIdImpl::kNoSegment; +} + +// Give each message plenty of time to arrive; a healthy broker delivers in milliseconds. +constexpr std::chrono::seconds kReceiveTimeout{20}; + +// A single-segment topic delivers in total order through one Exclusive consumer, and +// one cumulative ack of the last message settles the whole stream: a second consumer +// on the same subscription receives nothing. +TEST(StStreamConsumerE2ETest, testOrderedRoundTripAndCumulativeAck) { + if (!e2eEnabled()) GTEST_SKIP() << "set PULSAR_ST_E2E=1 to run against a scalable-topics broker"; + + const std::string name = uniqueName("st-e2e-stream"); + ASSERT_TRUE(createScalableTopic(name)) << "failed to create scalable topic " << name; + const std::string topic = topicUrl(name); + + auto clientResult = PulsarClient::builder().serviceUrl(serviceUrl()).build(); + ASSERT_TRUE(clientResult) << clientResult.error(); + PulsarClient client = std::move(clientResult).value(); + + auto consumerResult = client.newStreamConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(consumerResult) << consumerResult.error(); + StreamConsumer consumer = std::move(consumerResult).value(); + + auto producerResult = client.newProducer(Schema{}).topic(topic).create(); + ASSERT_TRUE(producerResult) << producerResult.error(); + Producer producer = std::move(producerResult).value(); + + constexpr int kCount = 25; + for (int i = 0; i < kCount; i++) { + auto sent = producer.newMessage().key("key-" + std::to_string(i % 4)).value("v-" + std::to_string(i)).send(); + ASSERT_TRUE(sent) << "send " << i << " failed: " << sent.error(); + } + ASSERT_TRUE(producer.flush()); + ASSERT_TRUE(producer.close()); + + // One segment, one Exclusive consumer: delivery is the publish order, exactly. + MessageId lastId = MessageId::earliest(); + for (int i = 0; i < kCount; i++) { + auto message = consumer.receive(kReceiveTimeout); + ASSERT_TRUE(message) << "receive " << i << " failed: " << message.error(); + EXPECT_EQ(message->value(), "v-" + std::to_string(i)) << "out of order at position " << i; + EXPECT_EQ(segmentIdOf(message->id()), 0); + lastId = message->id(); + } + // One cumulative ack of the last message settles everything delivered. + consumer.acknowledgeCumulative(lastId); + ASSERT_TRUE(consumer.close()); + + auto verifierResult = client.newStreamConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(verifierResult) << verifierResult.error(); + StreamConsumer verifier = std::move(verifierResult).value(); + auto redelivered = verifier.receive(std::chrono::seconds(3)); + ASSERT_FALSE(redelivered) << "cumulative ack did not stick: \"" << redelivered->value() + << "\" was redelivered"; + EXPECT_EQ(redelivered.error().result, pulsar::ResultTimeout); + EXPECT_TRUE(verifier.close()); + EXPECT_TRUE(client.close()); +} + +// The DAG-replay scenario: everything produced before a split sits in the sealed +// parent, everything after lands on the children — and the broker withholds the +// children from the assignment until the parent is drained FOR THIS SUBSCRIPTION. +// Acking as we go is what lets the parent's backlog reach zero and unblock the +// children; every parent message must arrive before any child message. +TEST(StStreamConsumerE2ETest, testDrainsSealedParentBeforeChildren) { + if (!e2eEnabled()) GTEST_SKIP() << "set PULSAR_ST_E2E=1 to run against a scalable-topics broker"; + + const std::string name = uniqueName("st-e2e-stream-replay"); + ASSERT_TRUE(createScalableTopic(name)) << "failed to create scalable topic " << name; + const std::string topic = topicUrl(name); + + auto clientResult = PulsarClient::builder().serviceUrl(serviceUrl()).build(); + ASSERT_TRUE(clientResult) << clientResult.error(); + PulsarClient client = std::move(clientResult).value(); + + // Create the durable subscription up front (and detach) so the pre-split backlog + // is retained for it. + { + auto subscriberResult = client.newStreamConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(subscriberResult) << subscriberResult.error(); + StreamConsumer subscriber = std::move(subscriberResult).value(); + ASSERT_TRUE(subscriber.close()); + } + + auto producerResult = client.newProducer(Schema{}).topic(topic).create(); + ASSERT_TRUE(producerResult) << producerResult.error(); + Producer producer = std::move(producerResult).value(); + + constexpr int kBefore = 60; + std::set producedBefore; + for (int i = 0; i < kBefore; i++) { + std::string value = "before-" + std::to_string(i); + auto sent = producer.newMessage().key("key-" + std::to_string(i)).value(value).send(); + ASSERT_TRUE(sent) << "pre-split send " << i << " failed: " << sent.error(); + producedBefore.insert(std::move(value)); + } + ASSERT_TRUE(producer.flush()); + + // Seal the parent with its backlog behind, then publish the post-split batch onto + // the children. + ASSERT_TRUE(splitSegment(name, 0)) << "failed to split segment 0 of " << name; + constexpr int kAfter = 40; + std::set producedAfter; + for (int i = 0; i < kAfter; i++) { + std::string value = "after-" + std::to_string(i); + auto sent = producer.newMessage().key("key-" + std::to_string(i)).value(value).send(); + ASSERT_TRUE(sent) << "post-split send " << i << " failed: " << sent.error(); + producedAfter.insert(std::move(value)); + } + ASSERT_TRUE(producer.flush()); + ASSERT_TRUE(producer.close()); + + auto consumerResult = client.newStreamConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(consumerResult) << consumerResult.error(); + StreamConsumer consumer = std::move(consumerResult).value(); + + // Ack cumulatively as we go: the broker decides the parent is drained by this + // subscription's backlog reaching zero, so deferring acks to the end would keep + // the children withheld forever. + std::set receivedBefore; + std::set receivedAfter; + bool sawChild = false; + for (int i = 0; i < kBefore + kAfter; i++) { + auto message = consumer.receive(kReceiveTimeout); + ASSERT_TRUE(message) << "receive " << i << " failed: " << message.error(); + if (segmentIdOf(message->id()) == 0) { + EXPECT_FALSE(sawChild) << "parent message \"" << message->value() + << "\" arrived after a child message — DAG order broken"; + receivedBefore.insert(std::string(message->value())); + } else { + sawChild = true; + receivedAfter.insert(std::string(message->value())); + } + consumer.acknowledgeCumulative(message->id()); + } + EXPECT_EQ(receivedBefore, producedBefore) << "the sealed parent's backlog did not fully arrive"; + EXPECT_EQ(receivedAfter, producedAfter) << "the post-split children's messages did not fully arrive"; + + EXPECT_TRUE(consumer.close()); + EXPECT_TRUE(client.close()); +} + +// Two initial segments (no parents, so both are assigned immediately): drain both via +// batch receives without acking, then acknowledge only the very last message — its +// position vector must advance BOTH segments' cursors. +TEST(StStreamConsumerE2ETest, testCumulativeAckCoversAllSegments) { + if (!e2eEnabled()) GTEST_SKIP() << "set PULSAR_ST_E2E=1 to run against a scalable-topics broker"; + + const std::string name = uniqueName("st-e2e-stream-vector"); + ASSERT_TRUE(createScalableTopic(name, /*numInitialSegments*/ 2)) << "failed to create " << name; + const std::string topic = topicUrl(name); + + auto clientResult = PulsarClient::builder().serviceUrl(serviceUrl()).build(); + ASSERT_TRUE(clientResult) << clientResult.error(); + PulsarClient client = std::move(clientResult).value(); + + auto consumerResult = client.newStreamConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(consumerResult) << consumerResult.error(); + StreamConsumer consumer = std::move(consumerResult).value(); + + auto producerResult = client.newProducer(Schema{}).topic(topic).create(); + ASSERT_TRUE(producerResult) << producerResult.error(); + Producer producer = std::move(producerResult).value(); + + // 60 distinct keys over two half-range segments hit both with overwhelming probability. + constexpr int kCount = 60; + std::set produced; + for (int i = 0; i < kCount; i++) { + std::string value = "v-" + std::to_string(i); + auto sent = producer.newMessage().key("key-" + std::to_string(i)).value(value).send(); + ASSERT_TRUE(sent) << "send " << i << " failed: " << sent.error(); + produced.insert(std::move(value)); + } + ASSERT_TRUE(producer.flush()); + ASSERT_TRUE(producer.close()); + + // Drain through batch receives, acking nothing along the way. + std::set received; + std::set segments; + MessageId lastId = MessageId::earliest(); + while (static_cast(received.size()) < kCount) { + auto batch = consumer.receiveMulti(20, std::chrono::seconds(5)); + ASSERT_TRUE(batch) << "batch receive failed: " << batch.error(); + ASSERT_FALSE(batch->empty()) << "drain stalled at " << received.size() << " of " << kCount; + for (const auto& message : *batch) { + segments.insert(segmentIdOf(message.id())); + received.insert(std::string(message.value())); + lastId = message.id(); + } + } + EXPECT_EQ(received, produced); + EXPECT_GE(segments.size(), 2u) << "messages did not arrive from both segments"; + + // One ack: its position vector advances every segment's cursor. + consumer.acknowledgeCumulative(lastId); + ASSERT_TRUE(consumer.close()); + + auto verifierResult = client.newStreamConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(verifierResult) << verifierResult.error(); + StreamConsumer verifier = std::move(verifierResult).value(); + auto redelivered = verifier.receive(std::chrono::seconds(3)); + ASSERT_FALSE(redelivered) << "the position vector did not cover every segment: \"" + << redelivered->value() << "\" was redelivered"; + EXPECT_EQ(redelivered.error().result, pulsar::ResultTimeout); + EXPECT_TRUE(verifier.close()); + EXPECT_TRUE(client.close()); +} + +} // namespace From b4609a1773008b1001e89ea3da2163b6715874d3 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 09:42:58 -0700 Subject: [PATCH 4/5] st: clang-format-11 line wrapping in the stream consumer sources --- lib/st/ConsumerAssignmentSession.cc | 18 ++++--- lib/st/ReceiveQueue.h | 3 +- lib/st/StreamConsumerImpl.cc | 73 ++++++++++++++--------------- lib/st/StreamConsumerImpl.h | 3 +- tests/st/StStreamConsumerE2ETest.cc | 7 +-- 5 files changed, 50 insertions(+), 54 deletions(-) diff --git a/lib/st/ConsumerAssignmentSession.cc b/lib/st/ConsumerAssignmentSession.cc index b90f5301..07cea3ed 100644 --- a/lib/st/ConsumerAssignmentSession.cc +++ b/lib/st/ConsumerAssignmentSession.cc @@ -115,15 +115,13 @@ void ConsumerAssignmentSession::connectAndSubscribe(detail::Promiseclient_->getServiceInfo().useTls(); - const auto& controllerUrl = - useTls ? result->controllerBrokerUrlTls() : result->controllerBrokerUrl(); + const auto& controllerUrl = useTls ? result->controllerBrokerUrlTls() : result->controllerBrokerUrl(); // Behind a proxy the controller's advertised address is not directly reachable, // and before leader election completes there is no address at all: in both // cases connect through the regular lookup path — any broker forwards the // subscribe to the controller and relays assignment updates back. - const bool useDirect = - controllerUrl.has_value() && !controllerUrl->empty() && - self->client_->getClientConfig().getProxyServiceUrl().empty(); + const bool useDirect = controllerUrl.has_value() && !controllerUrl->empty() && + self->client_->getClientConfig().getProxyServiceUrl().empty(); auto connectionFuture = useDirect ? self->client_->connect("", *controllerUrl, static_cast(self->consumerId_)) : self->client_->getConnection("", DagWatchSession::lookupCompatibleTopic(self->topic_), @@ -166,16 +164,16 @@ void ConsumerAssignmentSession::subscribeOn(const pulsar::ClientConnectionPtr& c } const std::uint64_t requestId = client_->newRequestId(); bool added = cnx->addScalableSubscribeRequest( - requestId, - [promise](pulsar::Result result, const pulsar::proto::CommandScalableTopicSubscribeResponse* response) { + requestId, [promise](pulsar::Result result, + const pulsar::proto::CommandScalableTopicSubscribeResponse* response) { if (result != pulsar::ResultOk) { promise.setError(Error{result, "connection closed before the subscribe response"}); return; } if (response->has_error()) { - promise.setError(Error{toSubscribeResult(response->error()), - response->has_message() ? response->message() - : "scalable-topic subscribe failed"}); + promise.setError( + Error{toSubscribeResult(response->error()), + response->has_message() ? response->message() : "scalable-topic subscribe failed"}); return; } if (!response->has_assignment()) { diff --git a/lib/st/ReceiveQueue.h b/lib/st/ReceiveQueue.h index 54cb62e3..d317d766 100644 --- a/lib/st/ReceiveQueue.h +++ b/lib/st/ReceiveQueue.h @@ -58,8 +58,7 @@ class ReceiveQueue : public std::enable_shared_from_this { * until the batch is full or the deadline elapses. May complete with fewer * than maxMessages — including zero on a quiet timeout. */ - Future> receiveMultiAsync(int maxMessages, - std::chrono::milliseconds timeout); + Future> receiveMultiAsync(int maxMessages, std::chrono::milliseconds timeout); /** Deliver a message; the returned future completes when there is room for the next offer. */ Future offer(MessageImplPtr message); diff --git a/lib/st/StreamConsumerImpl.cc b/lib/st/StreamConsumerImpl.cc index 5ad6ee9d..eb2effa7 100644 --- a/lib/st/StreamConsumerImpl.cc +++ b/lib/st/StreamConsumerImpl.cc @@ -92,9 +92,9 @@ Future StreamConsumerImpl::start() { "scalable-topics client"}); return startPromise_.getFuture(); } - session_ = std::make_shared( - classic_, config_.topic, config_.subscriptionName, consumerName_, - pulsar::ScalableConsumerType_STREAM); + session_ = + std::make_shared(classic_, config_.topic, config_.subscriptionName, + consumerName_, pulsar::ScalableConsumerType_STREAM); std::weak_ptr weak = weak_from_this(); session_->setListener([weak](const std::vector& newSegments, const std::vector& oldSegments) { @@ -259,40 +259,39 @@ bool StreamConsumerImpl::isSegmentStillAssignedLocked(std::uint64_t segmentId) c void StreamConsumerImpl::subscribeSegmentWithRetry(const AssignedSegment& assigned, int attempt) { std::weak_ptr weak = weak_from_this(); - getOrCreateSegmentConsumerAsync(assigned).addListener( - [weak, assigned, attempt](const Expected& result) { + getOrCreateSegmentConsumerAsync(assigned).addListener([weak, assigned, attempt]( + const Expected& result) { + auto self = weak.lock(); + if (result || !self || self->closed_.load()) return; + if (!isRebalanceCollision(result.error().result)) { + LOG_ERROR("[" << self->topic_ << "] segment " << assigned.segment.segmentId + << " subscribe failed (" << result.error() + << "); not a rebalance collision, waiting for the next assignment"); + return; + } + if (attempt + 1 >= kSubscribeRetryMaxAttempts) { + LOG_ERROR("[" << self->topic_ << "] segment " << assigned.segment.segmentId + << " subscribe still colliding after " << kSubscribeRetryMaxAttempts + << " attempts; giving up until the next assignment: " << result.error()); + return; + } + { + std::lock_guard lock(self->mutex_); + if (!self->isSegmentStillAssignedLocked(assigned.segment.segmentId)) return; + } + LOG_INFO("[" << self->topic_ << "] segment " << assigned.segment.segmentId + << " is still held by its previous owner; retrying, attempt " << (attempt + 1) << " of " + << kSubscribeRetryMaxAttempts); + auto timer = self->executor_->createDeadlineTimer(); + const std::int64_t delayMs = std::min(100 * (attempt + 1), kSubscribeRetryMaxBackoffMs); + timer->expires_from_now(std::chrono::milliseconds(delayMs)); + // Weak ref: closeAsync() does not cancel these timers (`timer` keeps itself alive). + timer->async_wait([weak, assigned, attempt, timer](const ASIO_ERROR& ec) { auto self = weak.lock(); - if (result || !self || self->closed_.load()) return; - if (!isRebalanceCollision(result.error().result)) { - LOG_ERROR("[" << self->topic_ << "] segment " << assigned.segment.segmentId - << " subscribe failed (" << result.error() - << "); not a rebalance collision, waiting for the next assignment"); - return; - } - if (attempt + 1 >= kSubscribeRetryMaxAttempts) { - LOG_ERROR("[" << self->topic_ << "] segment " << assigned.segment.segmentId - << " subscribe still colliding after " << kSubscribeRetryMaxAttempts - << " attempts; giving up until the next assignment: " << result.error()); - return; - } - { - std::lock_guard lock(self->mutex_); - if (!self->isSegmentStillAssignedLocked(assigned.segment.segmentId)) return; - } - LOG_INFO("[" << self->topic_ << "] segment " << assigned.segment.segmentId - << " is still held by its previous owner; retrying, attempt " << (attempt + 1) - << " of " << kSubscribeRetryMaxAttempts); - auto timer = self->executor_->createDeadlineTimer(); - const std::int64_t delayMs = - std::min(100 * (attempt + 1), kSubscribeRetryMaxBackoffMs); - timer->expires_from_now(std::chrono::milliseconds(delayMs)); - // Weak ref: closeAsync() does not cancel these timers (`timer` keeps itself alive). - timer->async_wait([weak, assigned, attempt, timer](const ASIO_ERROR& ec) { - auto self = weak.lock(); - if (ec || !self || self->closed_.load()) return; - self->subscribeSegmentWithRetry(assigned, attempt + 1); - }); + if (ec || !self || self->closed_.load()) return; + self->subscribeSegmentWithRetry(assigned, attempt + 1); }); + }); } void StreamConsumerImpl::startReceiveLoop(pulsar::Consumer consumer, std::uint64_t segmentId) { @@ -346,8 +345,8 @@ Future StreamConsumerImpl::receiveAsync(std::chrono::millisecond return receiveQueue_->receiveAsync(timeout); } -Future> StreamConsumerImpl::receiveMultiAsync( - int maxMessages, std::chrono::milliseconds timeout) { +Future> StreamConsumerImpl::receiveMultiAsync(int maxMessages, + std::chrono::milliseconds timeout) { return receiveQueue_->receiveMultiAsync(maxMessages, timeout); } diff --git a/lib/st/StreamConsumerImpl.h b/lib/st/StreamConsumerImpl.h index a6d4a12b..c4cafb26 100644 --- a/lib/st/StreamConsumerImpl.h +++ b/lib/st/StreamConsumerImpl.h @@ -73,8 +73,7 @@ class StreamConsumerImpl : public std::enable_shared_from_this receiveAsync(); Future receiveAsync(std::chrono::milliseconds timeout); - Future> receiveMultiAsync(int maxMessages, - std::chrono::milliseconds timeout); + Future> receiveMultiAsync(int maxMessages, std::chrono::milliseconds timeout); void acknowledgeCumulative(const MessageId& id); void acknowledgeCumulative(const MessageId& id, const Transaction& txn); Future closeAsync(); diff --git a/tests/st/StStreamConsumerE2ETest.cc b/tests/st/StStreamConsumerE2ETest.cc index fbd128cf..f056de24 100644 --- a/tests/st/StStreamConsumerE2ETest.cc +++ b/tests/st/StStreamConsumerE2ETest.cc @@ -74,7 +74,8 @@ TEST(StStreamConsumerE2ETest, testOrderedRoundTripAndCumulativeAck) { constexpr int kCount = 25; for (int i = 0; i < kCount; i++) { - auto sent = producer.newMessage().key("key-" + std::to_string(i % 4)).value("v-" + std::to_string(i)).send(); + auto sent = + producer.newMessage().key("key-" + std::to_string(i % 4)).value("v-" + std::to_string(i)).send(); ASSERT_TRUE(sent) << "send " << i << " failed: " << sent.error(); } ASSERT_TRUE(producer.flush()); @@ -266,8 +267,8 @@ TEST(StStreamConsumerE2ETest, testCumulativeAckCoversAllSegments) { ASSERT_TRUE(verifierResult) << verifierResult.error(); StreamConsumer verifier = std::move(verifierResult).value(); auto redelivered = verifier.receive(std::chrono::seconds(3)); - ASSERT_FALSE(redelivered) << "the position vector did not cover every segment: \"" - << redelivered->value() << "\" was redelivered"; + ASSERT_FALSE(redelivered) << "the position vector did not cover every segment: \"" << redelivered->value() + << "\" was redelivered"; EXPECT_EQ(redelivered.error().result, pulsar::ResultTimeout); EXPECT_TRUE(verifier.close()); EXPECT_TRUE(client.close()); From 5e9dfdf19fd8b5f4c45ec571e6fbf9162747b452 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 09:49:16 -0700 Subject: [PATCH 5/5] st: pass promise and batch handles by const reference in the session and mux queue clang-tidy performance-unnecessary-value-param (the Lint job's config): the connectAndSubscribe/subscribeOn/collectMulti parameters are shared-state handles only read (and copied into continuations) by the bodies, so take them by const reference instead of by value. --- lib/st/ConsumerAssignmentSession.cc | 4 ++-- lib/st/ConsumerAssignmentSession.h | 5 +++-- lib/st/ReceiveQueue.cc | 4 ++-- lib/st/ReceiveQueue.h | 4 ++-- 4 files changed, 9 insertions(+), 8 deletions(-) diff --git a/lib/st/ConsumerAssignmentSession.cc b/lib/st/ConsumerAssignmentSession.cc index 07cea3ed..e9e9fb2c 100644 --- a/lib/st/ConsumerAssignmentSession.cc +++ b/lib/st/ConsumerAssignmentSession.cc @@ -97,7 +97,7 @@ Future> ConsumerAssignmentSession::start() { return initialAssignmentPromise_.getFuture(); } -void ConsumerAssignmentSession::connectAndSubscribe(detail::Promise promise) { +void ConsumerAssignmentSession::connectAndSubscribe(const detail::Promise& promise) { // Resolve the controller leader through a one-shot DAG-watch lookup: scalable // topic URIs are not resolvable through the classic lookup service, and the // controller pushes assignment updates itself, so no long-lived layout watch is @@ -138,7 +138,7 @@ void ConsumerAssignmentSession::connectAndSubscribe(detail::Promise promise) { + const detail::Promise& promise) { if (closed_.load()) { promise.setError(Error{ResultAlreadyClosed, "consumer session closed"}); return; diff --git a/lib/st/ConsumerAssignmentSession.h b/lib/st/ConsumerAssignmentSession.h index d91c7085..afab16a1 100644 --- a/lib/st/ConsumerAssignmentSession.h +++ b/lib/st/ConsumerAssignmentSession.h @@ -104,8 +104,9 @@ class ConsumerAssignmentSession : public std::enable_shared_from_this promise); - void subscribeOn(const pulsar::ClientConnectionPtr& cnx, detail::Promise promise); + void connectAndSubscribe(const detail::Promise& promise); + void subscribeOn(const pulsar::ClientConnectionPtr& cnx, + const detail::Promise& promise); void handleSessionEvent(pulsar::Result result, const pulsar::proto::CommandScalableTopicAssignmentUpdate* update); // Epoch-gated apply + listener notification; used for the initial assignment, diff --git a/lib/st/ReceiveQueue.cc b/lib/st/ReceiveQueue.cc index 2e33f14b..a9fd830d 100644 --- a/lib/st/ReceiveQueue.cc +++ b/lib/st/ReceiveQueue.cc @@ -171,8 +171,8 @@ Future> ReceiveQueue::receiveMultiAsync(int maxMessa return promise.getFuture(); } -void ReceiveQueue::collectMulti(detail::Promise> promise, - std::shared_ptr> batch, int maxMessages, +void ReceiveQueue::collectMulti(const detail::Promise>& promise, + const std::shared_ptr>& batch, int maxMessages, std::chrono::steady_clock::time_point deadline) { std::deque> toSignal; bool closed = false; diff --git a/lib/st/ReceiveQueue.h b/lib/st/ReceiveQueue.h index d317d766..28d716ce 100644 --- a/lib/st/ReceiveQueue.h +++ b/lib/st/ReceiveQueue.h @@ -73,8 +73,8 @@ class ReceiveQueue : public std::enable_shared_from_this { // One receiveMultiAsync collection round: greedily drain what is buffered, then // wait for the next message with the remaining deadline and go again. - void collectMulti(detail::Promise> promise, - std::shared_ptr> batch, int maxMessages, + void collectMulti(const detail::Promise>& promise, + const std::shared_ptr>& batch, int maxMessages, std::chrono::steady_clock::time_point deadline); // A parked receive: the promise to complete and — for timed receives — the timeout timer,