diff --git a/include/pulsar/st/QueueConsumer.h b/include/pulsar/st/QueueConsumer.h index b7184466..94ea1bff 100644 --- a/include/pulsar/st/QueueConsumer.h +++ b/include/pulsar/st/QueueConsumer.h @@ -81,7 +81,8 @@ struct QueueConsumerConfig { * redelivery delay). Default-constructed `AckPolicy` when unset. */ AckPolicy ackPolicy; /** Optional dead-letter policy: route messages to a dead-letter topic after - * repeated redelivery. Default unset (no dead-lettering). */ + * repeated redelivery. Default unset (no dead-lettering). Not implemented yet: + * setting it fails the subscribe with `ResultOperationNotSupported`. */ std::optional deadLetterPolicy; /** Arbitrary client-side consumer properties (reported in topic stats). Default empty. */ Properties properties; @@ -328,6 +329,9 @@ class QueueConsumerBuilder { * Route messages to a dead-letter topic after repeated redelivery (spec §7.2). * QueueConsumer only. * + * Not implemented yet: setting a policy currently fails the subscribe with + * `ResultOperationNotSupported` rather than silently ignoring it. + * * @param policy the dead-letter policy (max redeliveries, DLQ topic name, etc.). * Default unset (no dead-lettering). * @return `*this` for chaining. diff --git a/include/pulsar/st/detail/QueueConsumerCore.h b/include/pulsar/st/detail/QueueConsumerCore.h index c67e24d3..0efaf366 100644 --- a/include/pulsar/st/detail/QueueConsumerCore.h +++ b/include/pulsar/st/detail/QueueConsumerCore.h @@ -33,6 +33,7 @@ namespace pulsar::st { class QueueConsumerImpl; using QueueConsumerImplPtr = std::shared_ptr; class Transaction; +class ClientImpl; // lib/st — mints consumer cores from subscribeQueueAsync namespace detail { @@ -60,6 +61,7 @@ class PULSAR_PUBLIC QueueConsumerCore { private: friend class ClientCore; + friend class ::pulsar::st::ClientImpl; explicit QueueConsumerCore(QueueConsumerImplPtr impl) : impl_(std::move(impl)) {} QueueConsumerImplPtr impl_; diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc index 474a96b1..39442ca5 100644 --- a/lib/ClientConnection.cc +++ b/lib/ClientConnection.cc @@ -853,6 +853,27 @@ void ClientConnection::handleActiveConsumerChange(const proto::CommandActiveCons } } +void ClientConnection::handleReachedEndOfTopic(const proto::CommandReachedEndOfTopic& reachedEndOfTopic) { + LOG_DEBUG(cnxString() << "Received reached-end-of-topic, consumer_id: " + << reachedEndOfTopic.consumer_id()); + Lock lock(mutex_); + ConsumersMap::iterator it = consumers_.find(reachedEndOfTopic.consumer_id()); + if (it != consumers_.end()) { + ConsumerImplPtr consumer = it->second.lock(); + if (consumer) { + lock.unlock(); + consumer->reachedEndOfTopic(); + } else { + consumers_.erase(reachedEndOfTopic.consumer_id()); + LOG_DEBUG(cnxString() << "Ignoring reached-end-of-topic for already destroyed consumer " + << reachedEndOfTopic.consumer_id()); + } + } else { + LOG_DEBUG(cnxString() << "Got invalid consumer Id in reached-end-of-topic " + << reachedEndOfTopic.consumer_id()); + } +} + void ClientConnection::handleIncomingMessage(const proto::CommandMessage& msg, bool isChecksumValid, proto::BrokerEntryMetadata& brokerEntryMetadata, proto::MessageMetadata& msgMetadata, SharedBuffer& payload) { @@ -997,6 +1018,10 @@ void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) { handleScalableTopicUpdate(incomingCmd.scalabletopicupdate()); break; + case BaseCommand::REACHED_END_OF_TOPIC: + handleReachedEndOfTopic(incomingCmd.reachedendoftopic()); + break; + default: LOG_WARN(cnxString() << "Received invalid message from server"); close(Error{ResultDisconnected, cnxString() + "Received invalid message from server"}); diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h index 8591f546..05f8578e 100644 --- a/lib/ClientConnection.h +++ b/lib/ClientConnection.h @@ -105,6 +105,7 @@ class CommandGetLastMessageIdResponse; class CommandLookupTopicResponse; class CommandPartitionedTopicMetadataResponse; class CommandProducerSuccess; +class CommandReachedEndOfTopic; class CommandScalableTopicUpdate; class CommandSendReceipt; class CommandSendError; @@ -265,6 +266,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this(callback, value); }); } +void ClientImpl::subscribeSegmentAsync(const std::string& topic, const std::string& subscriptionName, + const ConsumerConfiguration& conf, SubscribeV2Callback callback) { + subscribeToTopicsAsyncV2(topic, subscriptionName, conf, std::move(callback), + /* allowSegmentTopic */ true); +} + void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::string& subscriptionName, - const ConsumerConfiguration& conf, SubscribeV2Callback callback) { + const ConsumerConfiguration& conf, SubscribeV2Callback callback, + bool allowSegmentTopic) { LOG_INFO("Subscribing on Topic :" << topic); TopicNamePtr topicName; { @@ -627,7 +634,7 @@ void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::s } } - if (topicName->isSegment()) { + if (topicName->isSegment() && !allowSegmentTopic) { callback(segmentTopicRejected(topic)); return; } diff --git a/lib/ClientImpl.h b/lib/ClientImpl.h index 7b822c08..f7329284 100644 --- a/lib/ClientImpl.h +++ b/lib/ClientImpl.h @@ -103,6 +103,14 @@ class ClientImpl : public std::enable_shared_from_this { CreateProducerV2Callback callback, const std::optional& assignedBrokerUrl = std::nullopt); + /** + * Subscribe a consumer to a single `segment://` scalable-topic segment, bypassing the + * segment-domain rejection applied to the public subscribe path. The scalable-topics + * queue/stream consumers use this to attach a per-segment consumer. + */ + void subscribeSegmentAsync(const std::string& topic, const std::string& subscriptionName, + const ConsumerConfiguration& conf, SubscribeV2Callback callback); + void subscribeAsync(const std::string& topic, const std::string& subscriptionName, const ConsumerConfiguration& conf, const SubscribeCallback& callback); @@ -203,7 +211,8 @@ class ClientImpl : public std::enable_shared_from_this { ConsumerConfiguration conf, SubscribeV2Callback callback); void subscribeToTopicsAsyncV2(const std::string& topic, const std::string& subscriptionName, - const ConsumerConfiguration& conf, SubscribeV2Callback callback); + const ConsumerConfiguration& conf, SubscribeV2Callback callback, + bool allowSegmentTopic = false); void subscribeToTopicsAsyncV2(const std::vector& topics, const std::string& subscriptionName, const ConsumerConfiguration& conf, SubscribeV2Callback callback); diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index 657e1e60..47c93516 100644 --- a/lib/ConsumerImpl.cc +++ b/lib/ConsumerImpl.cc @@ -337,6 +337,11 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result incomingMessages_.clear(); possibleSendToDeadLetterTopicMessages_.clear(); backoff_.reset(); + // Re-derive end-of-topic from the new session: termination stops new publications, not + // redelivery of unacked messages, so a stale flag would report ResultTopicTerminated in + // the window before redeliveries arrive. The broker re-sends CommandReachedEndOfTopic + // once this consumer's read position reaches the terminate marker again. + hasReachedEndOfTopic_ = false; if (!messageListener_ && config_.getReceiverQueueSize() == 0) { // Complicated logic since we don't have a isLocked() function for mutex if (waitingForZeroQueueSizeMessage) { @@ -823,6 +828,24 @@ void ConsumerImpl::activeConsumerChanged(bool isActive) { } } +void ConsumerImpl::reachedEndOfTopic() { + hasReachedEndOfTopic_ = true; + // If nothing is buffered there is nothing left to deliver, so complete any waiting async + // receives with ResultTopicTerminated now. When messages are still buffered they drain through + // the normal path first, and the next receive observes the flag (see receiveAsync). + Lock lock(pendingReceiveMutex_); + if (incomingMessages_.empty()) { + Message msg; + while (!pendingReceives_.empty()) { + ReceiveCallback callback = pendingReceives_.front(); + pendingReceives_.pop(); + listenerExecutor_->postWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback, + get_shared_this_ptr(), ResultTopicTerminated, msg, + callback)); + } + } +} + void ConsumerImpl::internalConsumerChangeListener(bool isActive) { try { if (isActive) { @@ -1189,6 +1212,14 @@ void ConsumerImpl::receiveAsync(const ReceiveCallback& callback) { messageProcessed(msg); msg = interceptors_->beforeConsume(Consumer(shared_from_this()), msg); callback(ResultOk, msg); + } else if (hasReachedEndOfTopic_) { + // Terminated topic with nothing left buffered: fail the receive rather than parking it + // forever waiting for a message that will never arrive. + pendingReceiveMutexLock.unlock(); + if (config_.getReceiverQueueSize() == 0) { + mutexlock.unlock(); + } + callback(ResultTopicTerminated, msg); } else if (config_.getReceiverQueueSize() == 0) { pendingReceives_.push(callback); // If connection_ is nullptr, sendFlowPermitsToBroker does nothing. @@ -1217,6 +1248,13 @@ Result ConsumerImpl::receiveHelper(Message& msg) { return fetchSingleMessageFromBroker(msg); } + // A drained terminated topic has nothing left to deliver: fail fast instead of blocking + // forever, matching the async path. (A receive already parked in pop() when end-of-topic + // arrives still waits — the queue only wakes on a message or on close.) + if (hasReachedEndOfTopic_ && incomingMessages_.empty()) { + return ResultTopicTerminated; + } + if (!incomingMessages_.pop(msg)) { return ResultInterrupted; } @@ -1247,6 +1285,10 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) { return ResultInvalidConfiguration; } + if (hasReachedEndOfTopic_ && incomingMessages_.empty()) { + return ResultTopicTerminated; + } + if (incomingMessages_.pop(msg, std::chrono::milliseconds(timeout))) { messageProcessed(msg); msg = interceptors_->beforeConsume(Consumer(shared_from_this()), msg); @@ -1255,6 +1297,10 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) { if (state_ != Ready) { return ResultAlreadyClosed; } + // Waking up empty on a terminated topic means drained, not merely idle. + if (hasReachedEndOfTopic_ && incomingMessages_.empty()) { + return ResultTopicTerminated; + } return ResultTimeout; } } diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h index e2637624..ac6d934f 100644 --- a/lib/ConsumerImpl.h +++ b/lib/ConsumerImpl.h @@ -103,6 +103,9 @@ class ConsumerImpl : public ConsumerImplBase { proto::MessageMetadata& msgMetadata, SharedBuffer& payload); void messageProcessed(Message& msg, bool track = true); void activeConsumerChanged(bool isActive); + // The broker signalled that this (terminated) topic has no more messages beyond what has already + // been delivered. Surface ResultTopicTerminated to receivers once the prefetch queue drains. + void reachedEndOfTopic(); inline CommandSubscribe_SubType getSubType(); inline CommandSubscribe_InitialPosition getInitialPosition(); @@ -185,6 +188,10 @@ class ConsumerImpl : public ConsumerImplBase { private: std::atomic_bool waitingForZeroQueueSizeMessage; + // Set when the broker sends CommandReachedEndOfTopic and cleared again on each new broker + // session (termination does not cancel redelivery of unacked messages); a drained receive + // then yields ResultTopicTerminated instead of parking forever. + std::atomic_bool hasReachedEndOfTopic_{false}; std::shared_ptr get_shared_this_ptr(); bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData, const proto::MessageMetadata& metadata, SharedBuffer& payload, diff --git a/lib/st/MessageCore.cc b/lib/st/MessageCore.cc new file mode 100644 index 00000000..4a57a9dc --- /dev/null +++ b/lib/st/MessageCore.cc @@ -0,0 +1,38 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include + +#include "MessageImpl.h" + +namespace pulsar::st::detail { + +// Thin forwarders to the hidden MessageImpl (see ProducerCore.cc for the same pattern). +std::span MessageCore::data() const { return impl_->data(); } +MessageId MessageCore::id() const { return impl_->id(); } +std::optional MessageCore::key() const { return impl_->key(); } +const Properties& MessageCore::properties() const { return impl_->properties(); } +Timestamp MessageCore::publishTime() const { return impl_->publishTime(); } +std::optional MessageCore::eventTime() const { return impl_->eventTime(); } +int64_t MessageCore::sequenceId() const { return impl_->sequenceId(); } +std::optional MessageCore::producerName() const { return impl_->producerName(); } +std::string_view MessageCore::topic() const { return impl_->topic(); } +int MessageCore::redeliveryCount() const { return impl_->redeliveryCount(); } +std::optional MessageCore::replicatedFrom() const { return impl_->replicatedFrom(); } + +} // namespace pulsar::st::detail diff --git a/lib/st/MessageImpl.h b/lib/st/MessageImpl.h new file mode 100644 index 00000000..2751496d --- /dev/null +++ b/lib/st/MessageImpl.h @@ -0,0 +1,92 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#pragma once + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace pulsar::st { + +/** + * INTERNAL — the received message behind `detail::MessageCore`. + * + * A thin view over a classic `pulsar::Message` (which owns the payload and metadata) + * plus the segment-qualified `pulsar::st::MessageId` minted on the receive path. An + * optional `topicOverride` carries the scalable topic identity in namespace mode + * (a plain segment consumer reports the segment backing topic otherwise). + */ +class MessageImpl { + public: + MessageImpl(pulsar::Message message, MessageId id, + std::optional topicOverride = std::nullopt) + : classic_(std::move(message)), id_(std::move(id)), topicOverride_(std::move(topicOverride)) {} + + std::span data() const { + return {static_cast(classic_.getData()), classic_.getLength()}; + } + const MessageId& id() const { return id_; } + std::optional key() const { + if (!classic_.hasPartitionKey()) return std::nullopt; + return std::string_view(classic_.getPartitionKey()); + } + const Properties& properties() const { return classic_.getProperties(); } + Timestamp publishTime() const { return fromMillis(classic_.getPublishTimestamp()); } + std::optional eventTime() const { + const uint64_t millis = classic_.getEventTimestamp(); + return millis != 0 ? std::optional(fromMillis(millis)) : std::nullopt; + } + // The classic public Message API does not expose the message's sequence id; populating it + // would require reaching into pulsar::MessageImpl's metadata, i.e. touching the classic API. + // TODO: revisit when the Stream consumer needs it (a classic Message::getSequenceId() accessor). + int64_t sequenceId() const { return -1; } + std::optional producerName() const { + const std::string& name = classic_.getProducerName(); + return name.empty() ? std::nullopt : std::optional(name); + } + std::string_view topic() const { + return topicOverride_ ? std::string_view(*topicOverride_) : std::string_view(classic_.getTopicName()); + } + int redeliveryCount() const { return classic_.getRedeliveryCount(); } + std::optional replicatedFrom() const { + const std::optional from = classic_.getReplicatedFrom(); + if (!from || *from == nullptr) return std::nullopt; + return std::string_view(**from); + } + + private: + static Timestamp fromMillis(uint64_t millis) { + return Timestamp(std::chrono::milliseconds(static_cast(millis))); + } + + pulsar::Message classic_; + MessageId id_; + std::optional topicOverride_; +}; + +} // namespace pulsar::st diff --git a/lib/st/QueueConsumerCore.cc b/lib/st/QueueConsumerCore.cc new file mode 100644 index 00000000..6b4d51f3 --- /dev/null +++ b/lib/st/QueueConsumerCore.cc @@ -0,0 +1,47 @@ +/** + * 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 "QueueConsumerImpl.h" + +namespace pulsar::st::detail { + +// Thin forwarders to the hidden QueueConsumerImpl. The receive path maps the impl's MessageImplPtr +// to a MessageCore — the mapping lambda runs in this member context, which is a friend of +// MessageCore, so it can reach MessageCore's private constructor. +Future QueueConsumerCore::receiveAsync() const { + return impl_->receiveAsync().thenApply( + [](const MessageImplPtr& message) { return MessageCore{message}; }); +} +Future QueueConsumerCore::receiveAsync(std::chrono::milliseconds timeout) const { + return impl_->receiveAsync(timeout).thenApply( + [](const MessageImplPtr& message) { return MessageCore{message}; }); +} +void QueueConsumerCore::acknowledge(const MessageId& id) const { impl_->acknowledge(id); } +void QueueConsumerCore::acknowledge(const MessageId& id, const Transaction& txn) const { + impl_->acknowledge(id, txn); +} +void QueueConsumerCore::negativeAcknowledge(const MessageId& id) const { impl_->negativeAcknowledge(id); } +Future QueueConsumerCore::closeAsync() const { return impl_->closeAsync(); } +std::string_view QueueConsumerCore::topic() const { return impl_->topic(); } +std::string_view QueueConsumerCore::subscription() const { return impl_->subscription(); } +std::string_view QueueConsumerCore::consumerName() const { return impl_->consumerName(); } + +} // namespace pulsar::st::detail diff --git a/lib/st/QueueConsumerImpl.cc b/lib/st/QueueConsumerImpl.cc new file mode 100644 index 00000000..62604b09 --- /dev/null +++ b/lib/st/QueueConsumerImpl.cc @@ -0,0 +1,432 @@ +/** + * 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 "QueueConsumerImpl.h" + +#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) {}); + } + }); +} + +} // namespace + +QueueConsumerImpl::QueueConsumerImpl(pulsar::ClientImplPtr classic, QueueConsumerConfig config) + : classic_(std::move(classic)), + config_(std::move(config)), + topic_(config_.topic), + subscription_(config_.subscriptionName), + consumerName_(config_.consumerName.value_or(std::string{})), + executor_(classic_->getIOExecutorProvider()->get()), + receiveQueue_(std::make_shared(executor_, kReceiveQueueCapacity)), + currentLayout_(std::make_shared()) {} + +Future QueueConsumerImpl::start() { + if (config_.deadLetterPolicy) { + // Dead-lettering is not implemented yet: fail loudly rather than silently accepting a + // policy that would never fire. + startPromise_.setError(Error{ResultOperationNotSupported, + "deadLetterPolicy is not implemented yet in the scalable-topics " + "client; unset it to subscribe"}); + return startPromise_.getFuture(); + } + dagWatch_ = std::make_shared(classic_, config_.topic, /*createIfMissing*/ true); + std::weak_ptr weak = weak_from_this(); + dagWatch_->setLayoutChangeListener( + [weak](const SegmentLayout& newLayout, const SegmentLayout& oldLayout) { + if (auto self = weak.lock()) self->onLayoutChange(newLayout, oldLayout); + }); + dagWatch_->start().addListener([weak](const Expected& result) { + if (auto self = weak.lock()) self->onStartResult(result); + }); + return startPromise_.getFuture(); +} + +void QueueConsumerImpl::onStartResult(const Expected& result) { + // Only the failure path (see StProducerImpl::onStartResult): the layout listener drives the + // success path — subscribe the initial segments and complete startPromise_. + if (!result) startPromise_.setError(result.error()); +} + +void QueueConsumerImpl::onLayoutChange(const SegmentLayout& newLayout, const SegmentLayout& /*oldLayout*/) { + // Subscribe active AND sealed segments: a sealed segment may still hold undrained messages. + std::vector target; + target.reserve(newLayout.activeSegments().size() + newLayout.sealedSegments().size()); + for (const auto& segment : newLayout.activeSegments()) target.push_back(segment); + for (const auto& segment : newLayout.sealedSegments()) target.push_back(segment); + + std::vector> retired; + std::vector toAdd; + bool first = false; + { + std::lock_guard lock(mutex_); + first = !sawFirstLayout_; + sawFirstLayout_ = true; + currentLayout_ = std::make_shared(newLayout); + + std::unordered_set targetIds; + for (const auto& segment : target) targetIds.insert(segment.segmentId); + for (auto it = segmentConsumers_.begin(); it != segmentConsumers_.end();) { + if (targetIds.find(it->first) == targetIds.end()) { + retired.push_back(std::move(it->second)); + outstanding_.erase(it->first); + terminatedSegments_.erase(it->first); + it = segmentConsumers_.erase(it); + } else { + ++it; + } + } + // Forget segments that have left the DAG so a future segment id can never be mistaken for a + // previously-drained one. + for (auto it = drainedSegments_.begin(); it != drainedSegments_.end();) { + it = targetIds.count(*it) ? std::next(it) : drainedSegments_.erase(it); + } + for (const auto& segment : target) { + if (segmentConsumers_.find(segment.segmentId) == segmentConsumers_.end() && + drainedSegments_.find(segment.segmentId) == drainedSegments_.end()) + toAdd.push_back(segment); + } + } + + for (auto& future : retired) closeWhenReady(future); + + if (first) { + if (toAdd.empty()) { + startPromise_.setSuccess(); + return; + } + auto remaining = std::make_shared>(static_cast(toAdd.size())); + for (const auto& segment : toAdd) { + getOrCreateSegmentConsumerAsync(segment).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 { + // Off the start path no error can surface to a caller, so back a failed subscribe with a + // bounded retry — the DAG may stay quiet for a long time and the next push is the only + // other thing that would re-attempt the segment. + for (const auto& segment : toAdd) subscribeSegmentWithRetry(segment, /*attempt*/ 0); + } +} + +pulsar::ConsumerConfiguration QueueConsumerImpl::buildSegmentConfiguration(const Segment& segment) const { + // Build a FRESH config every time (pulsar::ConsumerConfiguration's copy ctor shares its impl). + pulsar::ConsumerConfiguration conf; + conf.setConsumerType(pulsar::ConsumerShared); + conf.setSchema(config_.schema); + conf.setSubscriptionInitialPosition(toClassicInitialPosition(config_.initialPosition)); + if (config_.consumerName) { + conf.setConsumerName(*config_.consumerName + "-seg-" + std::to_string(segment.segmentId)); + } + if (config_.ackPolicy.groupTime) { + conf.setAckGroupingTimeMs(static_cast(config_.ackPolicy.groupTime->count())); + } + if (config_.ackPolicy.negativeAckRedeliveryDelay) { + conf.setNegativeAckRedeliveryDelayMs( + static_cast(config_.ackPolicy.negativeAckRedeliveryDelay->count())); + } + for (const auto& [key, value] : config_.properties) conf.setProperty(key, value); + if (segment.isLegacy()) conf.setProperty("__pulsar.v5.managed", "true"); + return conf; +} + +Future QueueConsumerImpl::getOrCreateSegmentConsumerAsync(const Segment& segment) { + detail::Promise promise; + { + std::lock_guard lock(mutex_); + if (auto it = segmentConsumers_.find(segment.segmentId); it != segmentConsumers_.end()) { + return it->second; + } + segmentConsumers_.insert_or_assign(segment.segmentId, promise.getFuture()); + } + + const pulsar::ConsumerConfiguration conf = buildSegmentConfiguration(segment); + const std::string attachTopic = segment.attachTopicName(); + const std::uint64_t segmentId = segment.segmentId; + 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 reconcile retries this segment. + { + std::lock_guard lock(self->mutex_); + self->segmentConsumers_.erase(segmentId); + } + promise.setError(std::get(result)); + } + }); + return promise.getFuture(); +} + +bool QueueConsumerImpl::isSegmentStillWantedLocked(std::uint64_t segmentId) const { + if (drainedSegments_.count(segmentId) != 0) return false; + for (const auto& segment : currentLayout_->activeSegments()) { + if (segment.segmentId == segmentId) return true; + } + for (const auto& segment : currentLayout_->sealedSegments()) { + if (segment.segmentId == segmentId) return true; + } + return false; +} + +void QueueConsumerImpl::subscribeSegmentWithRetry(const Segment& segment, int attempt) { + std::weak_ptr weak = weak_from_this(); + getOrCreateSegmentConsumerAsync(segment).addListener([weak, segment, + attempt](const Expected& result) { + auto self = weak.lock(); + if (result || !self || self->closed_.load()) return; + if (attempt + 1 >= kSubscribeRetryMaxAttempts) { + LOG_ERROR("[" << self->topic_ << "] segment " << segment.segmentId << " subscribe failed after " + << kSubscribeRetryMaxAttempts + << " attempts; giving up until the next DAG update: " << result.error()); + return; + } + { + std::lock_guard lock(self->mutex_); + if (!self->isSegmentStillWantedLocked(segment.segmentId)) return; + } + LOG_WARN("[" << self->topic_ << "] segment " << segment.segmentId + << " subscribe failed; retrying, attempt " << (attempt + 1) << " of " + << kSubscribeRetryMaxAttempts << ": " << result.error()); + 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, so a strong one would keep the + // consumer alive until the backoff elapses. (`timer` keeps itself alive until it fires.) + timer->async_wait([weak, segment, attempt, timer](const ASIO_ERROR& ec) { + auto self = weak.lock(); + if (ec || !self || self->closed_.load()) return; + self->subscribeSegmentWithRetry(segment, attempt + 1); + }); + }); +} + +void QueueConsumerImpl::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's backlog is fully delivered — but end-of-topic only means + // the classic prefetch queue drained. Messages already fanned into the mux queue + // (or in the application's hands) still need this consumer to route their acks, so + // defer the close until every outstanding message settles (onMessageSettled + // finishes the drain then). Either way the segment is never re-subscribed. + std::optional> toClose; + { + std::lock_guard lock(self->mutex_); + auto outstanding = self->outstanding_.find(segmentId); + if (outstanding == self->outstanding_.end() || outstanding->second == 0) { + toClose = self->takeDrainedSegmentLocked(segmentId); + } else { + self->terminatedSegments_.insert(segmentId); + } + } + if (toClose) closeWhenReady(*toClose); + } + // Otherwise (AlreadyClosed / consumer closing) just stop the loop. + return; + } + MessageId id = MessageIdFactory::create(message.getMessageId(), static_cast(segmentId)); + // Report the scalable topic as the source, not the internal segment:// backing topic. + auto messageImpl = std::make_shared(message, std::move(id), self->topic_); + self->onMessageFannedIn(segmentId); + // Re-arm only once the fan-in queue has room, so a slow consumer throttles this segment — + // and hop through the executor rather than continuing inline: receiveAsync completes + // inline when a message is already prefetched and offer()'s future is already complete + // while the queue has room, so an inline continuation would recurse once per message and + // can exhaust the stack on a large backlog. + self->receiveQueue_->offer(std::move(messageImpl)) + .addListener([self, consumer, segmentId](const Expected&) { + self->executor_->postWork( + [self, consumer, segmentId] { self->startReceiveLoop(consumer, segmentId); }); + }); + }); +} + +void QueueConsumerImpl::onMessageFannedIn(std::uint64_t segmentId) { + std::lock_guard lock(mutex_); + ++outstanding_[segmentId]; +} + +void QueueConsumerImpl::onMessageSettled(std::uint64_t segmentId) { + std::optional> toClose; + { + std::lock_guard lock(mutex_); + auto it = outstanding_.find(segmentId); + if (it == outstanding_.end() || it->second == 0) return; // unknown segment or already balanced + if (--(it->second) == 0 && terminatedSegments_.count(segmentId) != 0) { + toClose = takeDrainedSegmentLocked(segmentId); + } + } + if (toClose) closeWhenReady(*toClose); +} + +std::optional> QueueConsumerImpl::takeDrainedSegmentLocked(std::uint64_t segmentId) { + std::optional> future; + if (auto it = segmentConsumers_.find(segmentId); it != segmentConsumers_.end()) { + future = std::move(it->second); + segmentConsumers_.erase(it); + } + terminatedSegments_.erase(segmentId); + outstanding_.erase(segmentId); + drainedSegments_.insert(segmentId); + return future; +} + +Future QueueConsumerImpl::receiveAsync() { return receiveQueue_->receiveAsync(); } + +Future QueueConsumerImpl::receiveAsync(std::chrono::milliseconds timeout) { + return receiveQueue_->receiveAsync(timeout); +} + +Future QueueConsumerImpl::segmentConsumerFor(const MessageId& id) const { + const auto& impl = MessageIdFactory::impl(id); + if (impl) { + std::lock_guard lock(mutex_); + if (auto it = segmentConsumers_.find(static_cast(impl->segmentId)); + it != segmentConsumers_.end()) { + return it->second; + } + } + detail::Promise promise; + promise.setError(Error{ResultUnknownError, "no consumer for the message's segment"}); + return promise.getFuture(); +} + +void QueueConsumerImpl::acknowledge(const MessageId& id) { + const auto& impl = MessageIdFactory::impl(id); + if (!impl) return; + const pulsar::MessageId v4 = impl->v4MessageId; + const auto segmentId = static_cast(impl->segmentId); + auto self = shared_from_this(); + segmentConsumerFor(id).addListener([self, v4, segmentId](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.acknowledgeAsync(v4, [](pulsar::Result) {}); + } + // Settle after the ack is enqueued, so a drain-deferred close still flushes it first. + self->onMessageSettled(segmentId); + }); +} + +void QueueConsumerImpl::acknowledge(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 message is simply redelivered. + LOG_WARN("[" << topic_ << "] transactional acknowledge is not implemented yet; dropping the ack"); +} + +void QueueConsumerImpl::negativeAcknowledge(const MessageId& id) { + const auto& impl = MessageIdFactory::impl(id); + if (!impl) return; + const pulsar::MessageId v4 = impl->v4MessageId; + const auto segmentId = static_cast(impl->segmentId); + auto self = shared_from_this(); + segmentConsumerFor(id).addListener([self, v4, segmentId](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.negativeAcknowledge(v4); + } + // A nack settles the message too: on a terminated segment its redelivery cannot reach this + // consumer again (the receive loop has ended), so the broker's cursor simply retains it + // for the subscription's next attach. + self->onMessageSettled(segmentId); + }); +} + +Future QueueConsumerImpl::closeAsync() { + if (closed_.exchange(true)) { + detail::Promise promise; + promise.setSuccess(); // idempotent + return promise.getFuture(); + } + if (dagWatch_) dagWatch_->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(); + outstanding_.clear(); + terminatedSegments_.clear(); + drainedSegments_.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/QueueConsumerImpl.h b/lib/st/QueueConsumerImpl.h new file mode 100644 index 00000000..d4a7c448 --- /dev/null +++ b/lib/st/QueueConsumerImpl.h @@ -0,0 +1,144 @@ +/** + * 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 + +#include "DagWatchSession.h" +#include "ReceiveQueue.h" +#include "SegmentLayout.h" +#include "lib/ClientImpl.h" +#include "lib/ExecutorService.h" + +namespace pulsar::st { + +/** + * The scalable-topics queue consumer (single scalable topic): a Shared subscription fanned across + * the topic's segments, a port of the Java v5 ScalableQueueConsumer. + * + * It owns one DagWatchSession and, per segment, a classic Shared-subscription pulsar::Consumer on + * that segment's segment:// backing topic (created via ClientImpl::subscribeSegmentAsync). Both + * active AND sealed segments are subscribed — a sealed segment may still hold undrained messages + * and pending acks. Each segment runs a receive loop that stamps the segment id onto every message + * and fans it into a shared ReceiveQueue; the user receives from that queue. Individual acks route + * back to the owning segment's consumer via the message id's segment id. Layout changes add + * consumers for new segments and close ones that left the DAG; a segment that reports + * TopicTerminated (a sealed segment fully drained) is closed and dropped — but only once every + * message it delivered has been acked or nacked, so acks for messages still in the mux queue or + * in the application's hands can still be routed. + */ +class QueueConsumerImpl : public std::enable_shared_from_this { + public: + QueueConsumerImpl(pulsar::ClientImplPtr classic, QueueConsumerConfig config); + + /** Start the DAG watch and subscribe the initial segments; completes once they are attached. */ + Future start(); + + Future receiveAsync(); + Future receiveAsync(std::chrono::milliseconds timeout); + void acknowledge(const MessageId& id); + void acknowledge(const MessageId& id, const Transaction& txn); + void negativeAcknowledge(const MessageId& id); + 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; + // Bounded retry for a segment subscribe that fails off the first-layout path (a layout push + // also retries, but the DAG may stay quiet for a long time). Mirrors the producer's constants. + static constexpr int kSubscribeRetryMaxAttempts = 10; + static constexpr std::int64_t kSubscribeRetryMaxBackoffMs = 500; + + pulsar::ConsumerConfiguration buildSegmentConfiguration(const Segment& segment) const; + Future getOrCreateSegmentConsumerAsync(const Segment& segment); + // getOrCreateSegmentConsumerAsync plus a bounded backoff retry on failure, used off the + // first-layout path where no start error surfaces the problem to the caller. + void subscribeSegmentWithRetry(const Segment& segment, int attempt); + void startReceiveLoop(pulsar::Consumer consumer, std::uint64_t segmentId); + + // Outstanding-message bookkeeping: fanned-in minus settled (acked or nacked). A terminated + // segment's consumer closes only once its count reaches zero, so late acks still route. + void onMessageFannedIn(std::uint64_t segmentId); + void onMessageSettled(std::uint64_t segmentId); + // Remove a fully-drained terminated segment's bookkeeping, mark it drained, and hand back its + // consumer future so the caller can close it outside the lock. Caller holds mutex_. + std::optional> takeDrainedSegmentLocked(std::uint64_t segmentId); + // Whether the segment is still in the current DAG and not already drained. Caller holds mutex_. + bool isSegmentStillWantedLocked(std::uint64_t segmentId) const; + + // start()'s future handler: surfaces a start-time lookup failure only; the success path (apply + // the initial layout, subscribe its segments, complete startPromise_) runs in the listener. + void onStartResult(const Expected& result); + void onLayoutChange(const SegmentLayout& newLayout, const SegmentLayout& oldLayout); + + // Route an ack/nack to the consumer that owns the message's segment; a no-op if that segment's + // consumer is gone (the message will simply be redelivered). + Future segmentConsumerFor(const MessageId& id) const; + + pulsar::ClientImplPtr classic_; + const QueueConsumerConfig config_; + const std::string topic_; + const std::string subscription_; + const std::string consumerName_; + // One IO executor shared with the ReceiveQueue: receive-loop re-arms hop through it (so the + // per-message chain is a loop, not recursion) and retry/timeout timers run on it. + const pulsar::ExecutorServicePtr executor_; + DagWatchSessionPtr dagWatch_; + ReceiveQueuePtr receiveQueue_; + detail::Promise startPromise_; + std::atomic closed_{false}; + + mutable std::mutex mutex_; + bool sawFirstLayout_ = false; // guarded by mutex_ + std::shared_ptr currentLayout_; // guarded by mutex_ + std::unordered_map> segmentConsumers_; // guarded by mutex_ + // Messages fanned in per segment that the application has not yet acked or nacked. Guarded by + // mutex_. + std::unordered_map outstanding_; + // Segments that reported end-of-topic while messages were still outstanding: their consumer + // stays in segmentConsumers_ for ack routing, and the close runs when the count hits zero. + // Guarded by mutex_. + std::unordered_set terminatedSegments_; + // Segments that have reported end-of-topic and been fully drained (closed); kept so a + // reconcile does not re-subscribe a still-in-DAG sealed segment (which would redeliver its + // unacked messages). Pruned when a segment leaves the DAG. Guarded by mutex_. + std::unordered_set drainedSegments_; +}; + +using QueueConsumerImplPtr = std::shared_ptr; + +} // namespace pulsar::st diff --git a/lib/st/ReceiveQueue.cc b/lib/st/ReceiveQueue.cc new file mode 100644 index 00000000..b8407a34 --- /dev/null +++ b/lib/st/ReceiveQueue.cc @@ -0,0 +1,183 @@ +/** + * 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 "ReceiveQueue.h" + +#include + +#include "MessageImpl.h" +#include "lib/ExecutorService.h" + +namespace pulsar::st { + +ReceiveQueue::ReceiveQueue(pulsar::ExecutorServicePtr executor, std::size_t capacity) + : executor_(std::move(executor)), capacity_(capacity) {} + +std::deque> ReceiveQueue::takeCapacityWaitersIfRoomLocked() { + std::deque> toSignal; + if (buffer_.size() < capacity_ && !capacityWaiters_.empty()) { + toSignal = std::move(capacityWaiters_); + capacityWaiters_.clear(); + } + return toSignal; +} + +Future ReceiveQueue::receiveAsync() { + detail::Promise promise; + MessageImplPtr message; + std::deque> toSignal; + { + std::lock_guard lock(mutex_); + if (closed_) { + promise.setError(Error{ResultAlreadyClosed, "consumer is closed"}); + return promise.getFuture(); + } + if (!buffer_.empty()) { + message = std::move(buffer_.front()); + buffer_.pop_front(); + toSignal = takeCapacityWaitersIfRoomLocked(); + } else { + pendingReceives_.emplace(nextReceiveId_++, PendingReceive{promise, nullptr}); + } + } + for (auto& waiter : toSignal) waiter.setSuccess(); + if (message) promise.setValue(std::move(message)); + return promise.getFuture(); +} + +Future ReceiveQueue::receiveAsync(std::chrono::milliseconds timeout) { + detail::Promise promise; + MessageImplPtr message; + std::deque> toSignal; + std::uint64_t receiveId = 0; + bool parked = false; + { + std::lock_guard lock(mutex_); + if (closed_) { + promise.setError(Error{ResultAlreadyClosed, "consumer is closed"}); + return promise.getFuture(); + } + if (!buffer_.empty()) { + message = std::move(buffer_.front()); + buffer_.pop_front(); + toSignal = takeCapacityWaitersIfRoomLocked(); + } else { + receiveId = nextReceiveId_++; + pendingReceives_.emplace(receiveId, PendingReceive{promise, nullptr}); + parked = true; + } + } + for (auto& waiter : toSignal) waiter.setSuccess(); + if (message) { + promise.setValue(std::move(message)); + return promise.getFuture(); + } + if (parked) { + // Attach the timer to the parked entry so delivery (or close) can cancel it — otherwise + // every timed receive would leave a live timer (holding this queue) until its deadline. + auto timer = executor_->createDeadlineTimer(); + bool armed = false; + { + std::lock_guard lock(mutex_); + auto it = pendingReceives_.find(receiveId); + if (it != pendingReceives_.end()) { + it->second.timer = timer; + armed = true; + } + } + // If a message (or close) already completed the receive, the timer is never started. + if (armed) { + timer->expires_from_now(timeout); + auto self = shared_from_this(); // keep the queue alive until the timer fires + timer->async_wait([self, receiveId, promise, timer](const ASIO_ERROR& ec) { + if (ec) return; // cancelled: a message (or close) won the race + { + std::lock_guard lock(self->mutex_); + auto it = self->pendingReceives_.find(receiveId); + if (it == self->pendingReceives_.end()) return; // a message was delivered first + self->pendingReceives_.erase(it); + } + promise.setError(Error{ResultTimeout, "receive timed out"}); + }); + } + } + return promise.getFuture(); +} + +Future ReceiveQueue::offer(MessageImplPtr message) { + detail::Promise receiver; + DeadlineTimerPtr receiverTimer; + bool deliver = false; + detail::Promise capacityPromise; + bool hasRoom = false; + { + std::lock_guard lock(mutex_); + if (closed_) { + hasRoom = true; + } else { + if (!pendingReceives_.empty()) { + auto oldest = pendingReceives_.begin(); // FIFO: lowest id + receiver = std::move(oldest->second.promise); + receiverTimer = std::move(oldest->second.timer); + pendingReceives_.erase(oldest); + deliver = true; + } else { + buffer_.push_back(std::move(message)); + } + if (buffer_.size() < capacity_) { + hasRoom = true; + } else { + capacityWaiters_.push_back(capacityPromise); + } + } + } + if (receiverTimer) { + ASIO_ERROR ignored; + receiverTimer->cancel(ignored); + } + if (deliver) receiver.setValue(std::move(message)); + if (hasRoom) { + detail::Promise ready; + ready.setSuccess(); + return ready.getFuture(); + } + return capacityPromise.getFuture(); +} + +void ReceiveQueue::close() { + std::map pending; + std::deque> waiters; + { + std::lock_guard lock(mutex_); + if (closed_) return; + closed_ = true; + pending.swap(pendingReceives_); + waiters.swap(capacityWaiters_); + buffer_.clear(); + } + for (auto& [id, entry] : pending) { + if (entry.timer) { + ASIO_ERROR ignored; + entry.timer->cancel(ignored); + } + entry.promise.setError(Error{ResultAlreadyClosed, "consumer is closed"}); + } + for (auto& waiter : waiters) waiter.setSuccess(); // let segment loops re-arm and see closed +} + +} // namespace pulsar::st diff --git a/lib/st/ReceiveQueue.h b/lib/st/ReceiveQueue.h new file mode 100644 index 00000000..762ffa0a --- /dev/null +++ b/lib/st/ReceiveQueue.h @@ -0,0 +1,85 @@ +/** + * 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 "lib/ExecutorService.h" + +namespace pulsar::st { + +/** + * The fan-in mux behind a queue/stream consumer: many per-segment receive loops `offer()` + * messages; the user `receiveAsync()`s them one at a time in FIFO order. + * + * Bounded to avoid unbounded buffering when the user is slow: `offer()` returns a future that + * completes only once the queue has room, and each segment loop awaits it before re-arming its + * own `receiveAsync()` — so a slow consumer throttles the underlying segment consumers' flow + * control rather than piling messages up in memory. + * + * A default (untimed) receive parks a promise until a message arrives; a timed receive fails that + * promise with `ResultTimeout` when the deadline elapses. `close()` fails every waiter. + */ +class ReceiveQueue : public std::enable_shared_from_this { + public: + ReceiveQueue(pulsar::ExecutorServicePtr executor, std::size_t capacity); + + Future receiveAsync(); + Future receiveAsync(std::chrono::milliseconds timeout); + + /** Deliver a message; the returned future completes when there is room for the next offer. */ + Future offer(MessageImplPtr message); + + /** Fail every pending receive (and release capacity waiters). Idempotent. */ + void close(); + + private: + // Signal capacity waiters if the buffer has drained below capacity. Caller holds mutex_; + // the returned promises must be completed after releasing it. + std::deque> takeCapacityWaitersIfRoomLocked(); + + // 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 { + detail::Promise promise; + DeadlineTimerPtr timer; + }; + + const pulsar::ExecutorServicePtr executor_; + const std::size_t capacity_; + + std::mutex mutex_; + std::deque buffer_; // guarded by mutex_ + std::map pendingReceives_; // guarded; FIFO by id + std::deque> capacityWaiters_; // guarded by mutex_ + std::uint64_t nextReceiveId_ = 0; // guarded by mutex_ + bool closed_ = false; // guarded by mutex_ +}; + +using ReceiveQueuePtr = std::shared_ptr; + +} // namespace pulsar::st diff --git a/lib/st/StClientImpl.cc b/lib/st/StClientImpl.cc index 40ec5b7e..1c6f4467 100644 --- a/lib/st/StClientImpl.cc +++ b/lib/st/StClientImpl.cc @@ -21,6 +21,7 @@ #include #include +#include "QueueConsumerImpl.h" #include "StProducerImpl.h" namespace pulsar::st { @@ -66,9 +67,18 @@ Future ClientImpl::subscribeStreamAsync(StreamConsum return notImplementedYet("subscribeStream"); } -// NOLINTNEXTLINE(performance-unnecessary-value-param) -Future ClientImpl::subscribeQueueAsync(QueueConsumerConfig) { - return notImplementedYet("subscribeQueue"); +Future ClientImpl::subscribeQueueAsync(QueueConsumerConfig 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::QueueConsumerCore{impl}); + } else { + promise.setError(result.error()); + } + }); + return promise.getFuture(); } // NOLINTNEXTLINE(performance-unnecessary-value-param) diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc index 5de5c755..99663d65 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -866,6 +866,63 @@ TEST(ConsumerTest, testIsConnected) { ASSERT_FALSE(consumer.isConnected()); } +// A consumer of a terminated topic drains the backlog and then reports ResultTopicTerminated on the +// async receive path, rather than dropping the connection (the pre-fix behaviour, which treated the +// broker's CommandReachedEndOfTopic as an invalid message) or parking the receive forever. +TEST(ConsumerTest, testReceiveAsyncAfterTopicTerminated) { + const std::string topicName = "testReceiveAsyncAfterTopicTerminated-" + std::to_string(time(nullptr)); + const std::string topic = "persistent://public/default/" + topicName; + + Client client(lookupUrl); + + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producer)); + + Consumer consumer; + ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", consumer)); + + constexpr int kCount = 5; + for (int i = 0; i < kCount; i++) { + ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent("m-" + std::to_string(i)).build())); + } + + const int httpCode = + makePostRequest(adminUrl + "admin/v2/persistent/public/default/" + topicName + "/terminate", ""); + ASSERT_EQ(200, httpCode) << "httpCode: " << httpCode; + + auto receiveWithin = [&consumer](std::chrono::seconds timeout, Message& out) { + auto promise = std::make_shared>>(); + consumer.receiveAsync([promise](Result result, const Message& msg) { + promise->set_value({result, msg}); + }); + auto future = promise->get_future(); + if (future.wait_for(timeout) != std::future_status::ready) return ResultTimeout; + auto pair = future.get(); + out = pair.second; + return pair.first; + }; + + // The backlog drains first... + for (int i = 0; i < kCount; i++) { + Message msg; + ASSERT_EQ(ResultOk, receiveWithin(std::chrono::seconds(10), msg)) << "message " << i; + ASSERT_EQ(ResultOk, consumer.acknowledge(msg)); + } + // ...then the terminated topic reports its end instead of hanging. + Message ignored; + ASSERT_EQ(ResultTopicTerminated, receiveWithin(std::chrono::seconds(10), ignored)); + + // The sync paths agree with the async path once the topic is drained: both the timed and the + // untimed receive fail fast with ResultTopicTerminated instead of waiting. + Message drained; + ASSERT_EQ(ResultTopicTerminated, consumer.receive(drained, 1000)); + ASSERT_EQ(ResultTopicTerminated, consumer.receive(drained)); + + ASSERT_EQ(ResultOk, consumer.close()); + ASSERT_EQ(ResultOk, producer.close()); + client.close(); +} + TEST(ConsumerTest, testPartitionsWithCloseUnblock) { Client client(lookupUrl); const std::string partitionedTopic = "testPartitionsWithCloseUnblock" + std::to_string(time(nullptr)); diff --git a/tests/st/StQueueConsumerE2ETest.cc b/tests/st/StQueueConsumerE2ETest.cc new file mode 100644 index 00000000..eef024fc --- /dev/null +++ b/tests/st/StQueueConsumerE2ETest.cc @@ -0,0 +1,299 @@ +/** + * 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 queue-consumer tests against a real scalable-topics broker: a produce -> consume +// round-trip over a Shared subscription. Gated on the PULSAR_ST_E2E environment variable so the +// ordinary (broker-free) unit-test run skips them; the docker harness sets it. The broker URL +// defaults to the standard test service and can be overridden with PULSAR_ST_E2E_SERVICE_URL. +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "lib/st/MessageIdImpl.h" +#include "tests/HttpHelper.h" + +using namespace pulsar::st; + +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); + 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}; + +TEST(StQueueConsumerE2ETest, testProduceThenConsumeRoundTrip) { + if (!e2eEnabled()) GTEST_SKIP() << "set PULSAR_ST_E2E=1 to run against a scalable-topics broker"; + + const std::string name = uniqueName("st-e2e-queue"); + 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(); + + // Subscribe first (Earliest) so the subscription and its per-segment cursors exist before we + // publish — every produced message is then guaranteed to be delivered. + auto consumerResult = client.newQueueConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(consumerResult) << consumerResult.error(); + QueueConsumer 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; + 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 % 4)).value(value).send(); + ASSERT_TRUE(sent) << "send " << i << " failed: " << sent.error(); + produced.insert(value); + } + ASSERT_TRUE(producer.flush()); + + // Receive exactly kCount messages; a Shared subscription gives no cross-segment order, so + // compare the received payloads as a set rather than a sequence. + std::set received; + for (int i = 0; i < kCount; i++) { + auto message = consumer.receive(kReceiveTimeout); + ASSERT_TRUE(message) << "receive " << i << " failed: " << message.error(); + EXPECT_GE(segmentIdOf(message->id()), 0) << "received message " << i << " has no real segment id"; + received.insert(message->value()); + consumer.acknowledge(message->id()); + } + EXPECT_EQ(received, produced) << "consumed payloads did not match what was produced"; + + EXPECT_TRUE(producer.close()); + EXPECT_TRUE(consumer.close()); + EXPECT_TRUE(client.close()); +} + +// Consume from a topic split (via REST) into two active segments and assert the messages actually +// arrive from both — this is the fan-in the queue consumer exists for: one Shared subscription +// multiplexed across a per-segment classic consumer each, drained through the mux receive queue. +// The single-segment round-trip above never exercises multi-segment fan-in. +TEST(StQueueConsumerE2ETest, testConsumeAcrossSplitSegments) { + if (!e2eEnabled()) GTEST_SKIP() << "set PULSAR_ST_E2E=1 to run against a scalable-topics broker"; + + const std::string name = uniqueName("st-e2e-queue-split"); + ASSERT_TRUE(createScalableTopic(name)) << "failed to create scalable topic " << name; + ASSERT_TRUE(splitSegment(name, 0)) << "failed to split segment 0 of " << 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.newQueueConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(consumerResult) << consumerResult.error(); + QueueConsumer 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(value); + } + ASSERT_TRUE(producer.flush()); + + std::set received; + std::set segments; + for (int i = 0; i < kCount; i++) { + auto message = consumer.receive(kReceiveTimeout); + ASSERT_TRUE(message) << "receive " << i << " failed: " << message.error(); + segments.insert(segmentIdOf(message->id())); + received.insert(message->value()); + consumer.acknowledge(message->id()); + } + EXPECT_EQ(received, produced) << "consumed payloads did not match what was produced"; + EXPECT_GE(segments.size(), 2u) << "messages did not fan in from both split segments"; + + EXPECT_TRUE(producer.close()); + EXPECT_TRUE(consumer.close()); + EXPECT_TRUE(client.close()); +} + +// The headline sealed-segment scenario: a split seals the parent WITHOUT migrating its backlog, so +// the messages produced before the split are only drainable through the sealed segment. Produce +// first, split, then consume: every pre-split message must still arrive (through the sealed +// parent), and the acks must stick — reattaching a second consumer on the same subscription gets +// nothing back. The count deliberately exceeds both the classic prefetch queue and the mux queue +// capacity (1000), so the drain also exercises back-pressure and the broker's end-of-topic +// arriving while messages are still unacked in the application's hands. +TEST(StQueueConsumerE2ETest, testDrainSealedSegmentBacklog) { + if (!e2eEnabled()) GTEST_SKIP() << "set PULSAR_ST_E2E=1 to run against a scalable-topics broker"; + + const std::string name = uniqueName("st-e2e-queue-drain"); + 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 backlog produced next is + // retained for it. + { + auto subscriberResult = client.newQueueConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(subscriberResult) << subscriberResult.error(); + QueueConsumer 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(); + + // Publish the backlog in bounded async waves (the per-segment pending-send queue is finite). + constexpr int kCount = 1200; + constexpr int kWave = 400; + std::set produced; + for (int base = 0; base < kCount; base += kWave) { + std::vector> wave; + wave.reserve(kWave); + for (int i = base; i < base + kWave; i++) { + std::string value = "v-" + std::to_string(i); + wave.push_back(producer.newMessage().key("key-" + std::to_string(i)).value(value).sendAsync()); + produced.insert(std::move(value)); + } + for (int i = 0; i < kWave; i++) { + auto sent = wave[i].get(); + ASSERT_TRUE(sent) << "send " << (base + i) << " failed: " << sent.error(); + } + } + ASSERT_TRUE(producer.flush()); + ASSERT_TRUE(producer.close()); + + // Seal the parent: its backlog stays behind in the sealed segment. + ASSERT_TRUE(splitSegment(name, 0)) << "failed to split segment 0 of " << name; + + // Drain the sealed segment through a fresh consumer, acking everything. + auto consumerResult = client.newQueueConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(consumerResult) << consumerResult.error(); + QueueConsumer consumer = std::move(consumerResult).value(); + + std::set received; + for (int i = 0; i < kCount; i++) { + auto message = consumer.receive(kReceiveTimeout); + ASSERT_TRUE(message) << "receive " << i << " failed: " << message.error(); + EXPECT_EQ(segmentIdOf(message->id()), 0) << "message " << i << " did not come from the sealed parent"; + received.insert(std::string(message->value())); + consumer.acknowledge(message->id()); + } + EXPECT_EQ(received, produced) << "the sealed segment's backlog did not drain completely"; + ASSERT_TRUE(consumer.close()); + + // The acks must have stuck: a second consumer on the same subscription gets nothing back. + auto verifierResult = client.newQueueConsumer(Schema{}) + .topic(topic) + .subscriptionName("sub") + .subscriptionInitialPosition(SubscriptionInitialPosition::Earliest) + .subscribe(); + ASSERT_TRUE(verifierResult) << verifierResult.error(); + QueueConsumer verifier = std::move(verifierResult).value(); + + auto redelivered = verifier.receive(std::chrono::seconds(3)); + ASSERT_FALSE(redelivered) << "acks were lost: message \"" << redelivered->value() + << "\" was redelivered after the drain"; + EXPECT_EQ(redelivered.error().result, pulsar::ResultTimeout); + + EXPECT_TRUE(verifier.close()); + EXPECT_TRUE(client.close()); +} + +} // namespace