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/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/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/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..e9e9fb2c --- /dev/null +++ b/lib/st/ConsumerAssignmentSession.cc @@ -0,0 +1,338 @@ +/** + * 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(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 + // 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, + const 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..afab16a1 --- /dev/null +++ b/lib/st/ConsumerAssignmentSession.h @@ -0,0 +1,140 @@ +/** + * 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(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, + // 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/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..a9fd830d 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(const detail::Promise>& promise, + const 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..28d716ce 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,14 @@ 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 +71,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(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, // 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..eb2effa7 --- /dev/null +++ b/lib/st/StreamConsumerImpl.cc @@ -0,0 +1,427 @@ +/** + * 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..c4cafb26 --- /dev/null +++ b/lib/st/StreamConsumerImpl.h @@ -0,0 +1,130 @@ +/** + * 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/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(); +} 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/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); +} diff --git a/tests/st/StStreamConsumerE2ETest.cc b/tests/st/StStreamConsumerE2ETest.cc new file mode 100644 index 00000000..f056de24 --- /dev/null +++ b/tests/st/StStreamConsumerE2ETest.cc @@ -0,0 +1,277 @@ +/** + * 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