From 779e2cfc25ca90d656ed040ddb33a2e7774db242 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 16 Jul 2026 11:41:56 -0700 Subject: [PATCH 01/12] =?UTF-8?q?st:=20received-message=20plumbing=20?= =?UTF-8?q?=E2=80=94=20MessageImpl=20+=20MessageCore=20accessors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consumer receive path needs the impl side of detail::MessageCore, which was declared but never defined (like ProducerCore was before the producer landed): - lib/st/MessageImpl.h: pulsar::st::MessageImpl, a thin view over a classic pulsar::Message (owns payload + metadata) plus the segment-qualified st MessageId minted on receive, with an optional topic override for namespace mode. - lib/st/MessageCore.cc: the out-of-line MessageCore accessors, forwarding to it. All accessors map to public classic Message getters except sequenceId(), which the classic public API does not expose; it returns -1 for now (a TODO to revisit with a classic accessor when the Stream consumer needs it, rather than touch the classic API here). Shared by all three consumer types. --- lib/st/MessageCore.cc | 38 ++++++++++++++++++ lib/st/MessageImpl.h | 91 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+) create mode 100644 lib/st/MessageCore.cc create mode 100644 lib/st/MessageImpl.h 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..7e9a4049 --- /dev/null +++ b/lib/st/MessageImpl.h @@ -0,0 +1,91 @@ +/** + * 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 From b0c7d111bd3c30c1ad9695821a4c13ce52f7027f Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 16 Jul 2026 11:45:07 -0700 Subject: [PATCH 02/12] =?UTF-8?q?st:=20classic=20consumer=20segment=20seam?= =?UTF-8?q?=20=E2=80=94=20subscribeSegmentAsync?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scalable-topics queue/stream consumers attach a Shared consumer per active segment, on the segment's segment:// backing topic — which the public subscribe path rejects. Add ClientImpl::subscribeSegmentAsync, mirroring the producer's createSegmentProducerAsync: the private single-topic subscribeToTopicsAsyncV2 gains an allowSegmentTopic flag (default false; the segment-domain rejection becomes isSegment() && !allowSegmentTopic), and the new public method calls it with true. A segment is a non-partitioned persistent topic, so it lands in the single-ConsumerImpl branch of handleSubscribe unchanged. No broker pin (the Java consumer path does not pin; the DAG-provided owner resolves via segment:// lookup). --- lib/ClientImpl.cc | 10 ++++++++-- lib/ClientImpl.h | 11 ++++++++++- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc index 510456f4..c9b0b28b 100644 --- a/lib/ClientImpl.cc +++ b/lib/ClientImpl.cc @@ -604,8 +604,14 @@ void ClientImpl::subscribeAsync(const std::string& topic, const std::string& sub [callback](const auto& value) { invokeLegacyCallback(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 +633,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); From f4f9ff50c0cba7888469c91542d8b85ed2a8b23d Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 16 Jul 2026 13:30:27 -0700 Subject: [PATCH 03/12] st: clang-format-11 line wrapping in segment seam + MessageImpl ctor Wrap two over-length lines that clang-format-11 (the CI style) breaks but clang-format-18 leaves on one line: the subscribeToTopicsAsyncV2 call in ClientImpl::subscribeSegmentAsync and the MessageImpl constructor signature. Formatting only, no behavior change. --- lib/ClientImpl.cc | 3 ++- lib/st/MessageImpl.h | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc index c9b0b28b..043d6d42 100644 --- a/lib/ClientImpl.cc +++ b/lib/ClientImpl.cc @@ -606,7 +606,8 @@ void ClientImpl::subscribeAsync(const std::string& topic, const std::string& sub 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); + subscribeToTopicsAsyncV2(topic, subscriptionName, conf, std::move(callback), + /* allowSegmentTopic */ true); } void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::string& subscriptionName, diff --git a/lib/st/MessageImpl.h b/lib/st/MessageImpl.h index 7e9a4049..2751496d 100644 --- a/lib/st/MessageImpl.h +++ b/lib/st/MessageImpl.h @@ -43,7 +43,8 @@ namespace pulsar::st { */ class MessageImpl { public: - MessageImpl(pulsar::Message message, MessageId id, std::optional topicOverride = std::nullopt) + 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 { From e110361e830e1828cd43294f48cd7fedf5cd12e1 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 16 Jul 2026 13:30:39 -0700 Subject: [PATCH 04/12] =?UTF-8?q?st:=20queue=20consumer=20core=20=E2=80=94?= =?UTF-8?q?=20per-segment=20fan-in=20over=20a=20mux=20receive=20queue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the single-topic scalable-topics queue consumer (a port of the Java v5 ScalableQueueConsumer). A Shared subscription is fanned across every segment of the topic — active AND sealed, since a sealed segment may still hold undrained messages — with one classic Shared-subscription pulsar::Consumer per segment created through the ClientImpl::subscribeSegmentAsync seam. - ReceiveQueue: a bounded fan-in mux. Per-segment receive loops offer() messages; the user receiveAsync()es them in FIFO order. offer() returns a future that completes only when the queue has room, so a slow consumer back-pressures the underlying segment consumers' flow control rather than buffering unboundedly. Timed receives fail with ResultTimeout; close() fails every waiter. - QueueConsumerImpl: owns one DagWatchSession and the per-segment consumers. Each segment loop stamps the segment id onto every message (MessageIdFactory) and fans it into the shared queue. Acks/nacks 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 drained sealed segment) is closed and dropped. - QueueConsumerCore: thin forwarders mapping MessageImplPtr to MessageCore. - Wire ClientImpl::subscribeQueueAsync (was notImplementedYet), mirroring createProducerAsync: build the impl, start(), then mint the public core. Transactional acknowledge is not implemented yet (logged and dropped); the dead-letter and namespace-subscription paths are deferred to later slices. --- include/pulsar/st/detail/QueueConsumerCore.h | 2 + lib/st/QueueConsumerCore.cc | 47 +++ lib/st/QueueConsumerImpl.cc | 303 +++++++++++++++++++ lib/st/QueueConsumerImpl.h | 107 +++++++ lib/st/ReceiveQueue.cc | 157 ++++++++++ lib/st/ReceiveQueue.h | 81 +++++ lib/st/StClientImpl.cc | 16 +- 7 files changed, 710 insertions(+), 3 deletions(-) create mode 100644 lib/st/QueueConsumerCore.cc create mode 100644 lib/st/QueueConsumerImpl.cc create mode 100644 lib/st/QueueConsumerImpl.h create mode 100644 lib/st/ReceiveQueue.cc create mode 100644 lib/st/ReceiveQueue.h 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/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..ea27e0ca --- /dev/null +++ b/lib/st/QueueConsumerImpl.cc @@ -0,0 +1,303 @@ +/** + * 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 "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; +} + +} // 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{})), + receiveQueue_( + std::make_shared(classic_->getIOExecutorProvider()->get(), kReceiveQueueCapacity)), + currentLayout_(std::make_shared()) {} + +Future QueueConsumerImpl::start() { + 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)); + it = segmentConsumers_.erase(it); + } else { + ++it; + } + } + for (const auto& segment : target) { + if (segmentConsumers_.find(segment.segmentId) == segmentConsumers_.end()) + toAdd.push_back(segment); + } + } + + for (auto& future : retired) { + future.addListener([](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.closeAsync([](pulsar::Result) {}); + } + }); + } + + 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 { + for (const auto& segment : toAdd) getOrCreateSegmentConsumerAsync(segment); // best-effort + } +} + +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 c = std::move(*consumer); + self->startReceiveLoop(c, segmentId); + promise.setValue(std::move(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(); +} + +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) { + // A sealed segment fully drained: close its consumer and drop it from the cache. + { + std::lock_guard lock(self->mutex_); + self->segmentConsumers_.erase(segmentId); + } + pulsar::Consumer done = consumer; + done.closeAsync([](pulsar::Result) {}); + } + // Otherwise (AlreadyClosed / consumer closing) just stop the loop. + return; + } + MessageId id = MessageIdFactory::create(message.getMessageId(), static_cast(segmentId)); + auto messageImpl = std::make_shared(message, std::move(id)); + // Re-arm only once the fan-in queue has room, so a slow consumer throttles this segment. + self->receiveQueue_->offer(std::move(messageImpl)) + .addListener([self, consumer, segmentId](const Expected&) { + self->startReceiveLoop(consumer, segmentId); + }); + }); +} + +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; + segmentConsumerFor(id).addListener([v4](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.acknowledgeAsync(v4, [](pulsar::Result) {}); + } + }); +} + +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; + segmentConsumerFor(id).addListener([v4](const Expected& result) { + if (result) { + pulsar::Consumer consumer = *result; + consumer.negativeAcknowledge(v4); + } + }); +} + +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(); + } + + 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..9065a729 --- /dev/null +++ b/lib/st/QueueConsumerImpl.h @@ -0,0 +1,107 @@ +/** + * 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 "DagWatchSession.h" +#include "ReceiveQueue.h" +#include "SegmentLayout.h" +#include "lib/ClientImpl.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. + */ +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; + + pulsar::ConsumerConfiguration buildSegmentConfiguration(const Segment& segment) const; + Future getOrCreateSegmentConsumerAsync(const Segment& segment); + void startReceiveLoop(pulsar::Consumer consumer, std::uint64_t segmentId); + + // 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_; + 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_ +}; + +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..76290923 --- /dev/null +++ b/lib/st/ReceiveQueue.cc @@ -0,0 +1,157 @@ +/** + * 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_++, promise); + } + } + 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, promise); + parked = true; + } + } + for (auto& waiter : toSignal) waiter.setSuccess(); + if (message) { + promise.setValue(std::move(message)); + return promise.getFuture(); + } + if (parked) { + auto timer = executor_->createDeadlineTimer(); + 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 + { + 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; + 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); + pendingReceives_.erase(oldest); + deliver = true; + } else { + buffer_.push_back(std::move(message)); + } + if (buffer_.size() < capacity_) { + hasRoom = true; + } else { + capacityWaiters_.push_back(capacityPromise); + } + } + } + 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, promise] : pending) 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..fe071d15 --- /dev/null +++ b/lib/st/ReceiveQueue.h @@ -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. + */ +#pragma once + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace pulsar { +class ExecutorService; +using ExecutorServicePtr = std::shared_ptr; +} // namespace pulsar + +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(); + + 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) From 6b3285ecc59b86717b9f1cbc60f2d15d0325255b Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 16 Jul 2026 15:52:40 -0700 Subject: [PATCH 05/12] Handle CommandReachedEndOfTopic on the consumer receive path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The classic client never handled BaseCommand::REACHED_END_OF_TOPIC (type 27): handleIncomingCommand fell through to default: and closed the whole connection as an "invalid message from server". Any consumer of a terminated topic — and every scalable-topics queue consumer, which subscribes to sealed segments to drain their backlog — would therefore churn its connection (close, reconnect, re-subscribe, reach end of topic again) instead of learning the topic ended. Handle it: dispatch REACHED_END_OF_TOPIC to the target consumer (mirroring handleActiveConsumerChange), and have ConsumerImpl surface ResultTopicTerminated on the async receive path once the prefetch queue drains — matching the Java client, whose consumers close a drained sealed segment on TopicTerminated. The broker only sends the command once the consumer's read position reaches the terminate marker, so buffered messages always drain before termination. Scope is the async receiveAsync path (what the scalable consumer uses); the blocking sync receive() is unchanged (it would need to interrupt a parked pop(), and a terminated topic there already behaves as "no more messages"). Adds ConsumerTest.testReceiveAsyncAfterTopicTerminated. --- lib/ClientConnection.cc | 25 ++++++++++++++++++++ lib/ClientConnection.h | 2 ++ lib/ConsumerImpl.cc | 26 +++++++++++++++++++++ lib/ConsumerImpl.h | 6 +++++ tests/ConsumerTest.cc | 51 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 110 insertions(+) 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_thispostWork(std::bind(&ConsumerImpl::notifyPendingReceivedCallback, + get_shared_this_ptr(), ResultTopicTerminated, msg, + callback)); + } + } +} + void ConsumerImpl::internalConsumerChangeListener(bool isActive) { try { if (isActive) { @@ -1189,6 +1207,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. diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h index e2637624..60973051 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,9 @@ class ConsumerImpl : public ConsumerImplBase { private: std::atomic_bool waitingForZeroQueueSizeMessage; + // Set once the broker sends CommandReachedEndOfTopic; 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/tests/ConsumerTest.cc b/tests/ConsumerTest.cc index 5de5c755..5fd22120 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -866,6 +866,57 @@ 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)); + + 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)); From 74757908acd32f5959a5759f8446d8710e23e054 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Thu, 16 Jul 2026 15:52:59 -0700 Subject: [PATCH 06/12] st: queue consumer produce->consume e2e test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end coverage for the scalable-topics queue consumer against a real broker, gated on PULSAR_ST_E2E (the broker-free unit run skips it): - testProduceThenConsumeRoundTrip: produce 25 keyed messages, receive and ack all of them through a Shared subscription, assert the payloads round-trip and every received id carries a real segment id. - testConsumeAcrossSplitSegments: over a topic pre-split into two active segments, produce 60 keyed messages and assert they fan in from both segments through the mux receive queue — the multi-segment path the queue consumer exists for, and the case that exercises draining the sealed parent segment. Both pass against apachepulsar/pulsar:5.0.0-M1. The CI wiring (docker-compose + run-unit-tests.sh) that runs these lands with the producer-e2e harness. --- tests/st/StQueueConsumerE2ETest.cc | 156 +++++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 tests/st/StQueueConsumerE2ETest.cc diff --git a/tests/st/StQueueConsumerE2ETest.cc b/tests/st/StQueueConsumerE2ETest.cc new file mode 100644 index 00000000..9bf3b03f --- /dev/null +++ b/tests/st/StQueueConsumerE2ETest.cc @@ -0,0 +1,156 @@ +/** + * 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 "lib/st/MessageIdImpl.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"; +} + +// 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 topic = "topic://public/default/st-e2e-queue"; + + 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 the harness has split 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 topic = "topic://public/default/st-e2e-queue-split"; + + 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()); +} + +} // namespace From acaddbfd19e4966d1ca805ce68361b2729f3975e Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 20 Jul 2026 08:45:07 -0700 Subject: [PATCH 07/12] st: drive queue-consumer e2e split through the admin REST API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror the producer e2e (#603): each queue-consumer e2e test now creates its own fresh-named scalable topic — and, for the fan-in test, splits it — through the admin REST API, instead of consuming harness-pre-created, CLI-pre-split fixed topics (st-e2e-queue / st-e2e-queue-split). The tests are now self-contained and keep working under the REST-driven harness, where nothing is pre-arranged for them. Links HttpHelper.cc into pulsar-st-tests for the makePut/makePostRequest calls. --- tests/BuildTests.cmake | 2 +- tests/st/StQueueConsumerE2ETest.cc | 59 +++++++++++++++++++++++++++--- 2 files changed, 54 insertions(+), 7 deletions(-) diff --git a/tests/BuildTests.cmake b/tests/BuildTests.cmake index 45dd84f4..c018a5bb 100644 --- a/tests/BuildTests.cmake +++ b/tests/BuildTests.cmake @@ -63,7 +63,7 @@ target_link_libraries(ExtensibleLoadManagerTest PRIVATE pulsarStatic ${GTEST_TAR # Pure client-side tests for the st API and its lib/st implementation; they do # not require a running broker. C++20 per-target, like the st API itself. file(GLOB ST_TEST_SOURCES st/*.cc) -add_executable(pulsar-st-tests ${ST_TEST_SOURCES}) +add_executable(pulsar-st-tests ${ST_TEST_SOURCES} HttpHelper.cc) set_target_properties(pulsar-st-tests PROPERTIES CXX_STANDARD 20 CXX_STANDARD_REQUIRED ON) target_include_directories(pulsar-st-tests PRIVATE ${AUTOGEN_DIR}/lib) target_link_libraries(pulsar-st-tests PRIVATE pulsarStatic ${GTEST_TARGETS}) diff --git a/tests/st/StQueueConsumerE2ETest.cc b/tests/st/StQueueConsumerE2ETest.cc index 9bf3b03f..67771df1 100644 --- a/tests/st/StQueueConsumerE2ETest.cc +++ b/tests/st/StQueueConsumerE2ETest.cc @@ -26,11 +26,14 @@ #include #include #include +#include #include #include +#include #include #include "lib/st/MessageIdImpl.h" +#include "tests/HttpHelper.h" using namespace pulsar::st; @@ -43,6 +46,43 @@ std::string serviceUrl() { 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); @@ -54,7 +94,10 @@ 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 topic = "topic://public/default/st-e2e-queue"; + + 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(); @@ -101,13 +144,17 @@ TEST(StQueueConsumerE2ETest, testProduceThenConsumeRoundTrip) { EXPECT_TRUE(client.close()); } -// Consume from a topic the harness has split 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. +// 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 topic = "topic://public/default/st-e2e-queue-split"; + + 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(); From eec4aac4ff048fb2df7477b179da2573b155d093 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 20 Jul 2026 14:43:31 -0700 Subject: [PATCH 08/12] =?UTF-8?q?st:=20address=20#605=20review=20=E2=80=94?= =?UTF-8?q?=20clang-tidy=20move=20+=20drained-segment=20re-subscribe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The queue-consumer subscribe callback applied std::move to a pulsar::Consumer, whose virtual destructor suppresses the move constructor, so the move bound to the copy constructor: clang-tidy performance-move-const-arg, which failed the Lint job. Copy the handle directly (a shared-impl copy), matching StProducerImpl. - On ResultTopicTerminated the drained segment's consumer was erased, but the sealed segment stays in the DAG, so the next layout reconcile re-subscribed it and the broker redelivered its still-unacked messages as duplicates. Track drained segments and skip re-subscribing them; prune the set when a segment leaves the DAG. --- lib/st/QueueConsumerImpl.cc | 19 +++++++++++++++---- lib/st/QueueConsumerImpl.h | 5 +++++ 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/lib/st/QueueConsumerImpl.cc b/lib/st/QueueConsumerImpl.cc index ea27e0ca..695b2964 100644 --- a/lib/st/QueueConsumerImpl.cc +++ b/lib/st/QueueConsumerImpl.cc @@ -97,8 +97,14 @@ void QueueConsumerImpl::onLayoutChange(const SegmentLayout& newLayout, const Seg ++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()) + if (segmentConsumers_.find(segment.segmentId) == segmentConsumers_.end() && + drainedSegments_.find(segment.segmentId) == drainedSegments_.end()) toAdd.push_back(segment); } } @@ -172,9 +178,11 @@ Future QueueConsumerImpl::getOrCreateSegmentConsumerAsync(cons attachTopic, config_.subscriptionName, conf, [self, promise, segmentId](std::variant result) { if (auto* consumer = std::get_if(&result)) { - pulsar::Consumer c = std::move(*consumer); + // 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(std::move(c)); + promise.setValue(c); } else { // Evict the failed subscribe so a later reconcile retries this segment. { @@ -193,10 +201,13 @@ void QueueConsumerImpl::startReceiveLoop(pulsar::Consumer consumer, std::uint64_ consumer.receiveAsync([self, consumer, segmentId](pulsar::Result result, const pulsar::Message& message) { if (result != pulsar::ResultOk) { if (result == pulsar::ResultTopicTerminated) { - // A sealed segment fully drained: close its consumer and drop it from the cache. + // A sealed segment fully drained: close its consumer, drop it from the cache, and + // remember it drained so a later reconcile does not re-subscribe the still-in-DAG + // sealed segment (which would redeliver its unacked messages as duplicates). { std::lock_guard lock(self->mutex_); self->segmentConsumers_.erase(segmentId); + self->drainedSegments_.insert(segmentId); } pulsar::Consumer done = consumer; done.closeAsync([](pulsar::Result) {}); diff --git a/lib/st/QueueConsumerImpl.h b/lib/st/QueueConsumerImpl.h index 9065a729..a79cbc8c 100644 --- a/lib/st/QueueConsumerImpl.h +++ b/lib/st/QueueConsumerImpl.h @@ -30,6 +30,7 @@ #include #include #include +#include #include "DagWatchSession.h" #include "ReceiveQueue.h" @@ -100,6 +101,10 @@ class QueueConsumerImpl : public std::enable_shared_from_this bool sawFirstLayout_ = false; // guarded by mutex_ std::shared_ptr currentLayout_; // guarded by mutex_ std::unordered_map> segmentConsumers_; // guarded by mutex_ + // Segments that have reported end-of-topic and been drained; 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; From cce4b820f31cd91ff2f1abd89170f287a5c9de88 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 12:17:17 -0700 Subject: [PATCH 09/12] Terminated-topic completeness on the consumer: reconnect + sync receive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps in the CommandReachedEndOfTopic handling, from #605 review: - hasReachedEndOfTopic_ was never cleared, so after a reconnect (which clears the prefetch queue and re-sends flow permits) a receiveAsync landing before redelivery arrived would report a stale ResultTopicTerminated — and the scalable queue consumer would then drop the segment permanently. Termination stops new publications, not redelivery of unacked messages: clear the flag on each new broker session; the broker re-sends the command once the re-created consumer's read position reaches the terminate marker again. - The sync receive paths ignored the flag entirely: the untimed receive() blocked forever on a drained terminated topic and the timed one returned ResultTimeout. Both now fail fast with ResultTopicTerminated when the flag is set and the queue is empty, agreeing with the async path. (A receive already parked in pop() when the command arrives still waits — waking it would need an interruptible queue.) Extends ConsumerTest.testReceiveAsyncAfterTopicTerminated to assert both sync overloads. --- lib/ConsumerImpl.cc | 20 ++++++++++++++++++++ lib/ConsumerImpl.h | 5 +++-- tests/ConsumerTest.cc | 6 ++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc index ac619de2..ebe95b82 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) { @@ -1243,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; } @@ -1273,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); @@ -1281,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 60973051..ac6d934f 100644 --- a/lib/ConsumerImpl.h +++ b/lib/ConsumerImpl.h @@ -188,8 +188,9 @@ class ConsumerImpl : public ConsumerImplBase { private: std::atomic_bool waitingForZeroQueueSizeMessage; - // Set once the broker sends CommandReachedEndOfTopic; a drained receive then yields - // ResultTopicTerminated instead of parking forever. + // 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, diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc index 5fd22120..99663d65 100644 --- a/tests/ConsumerTest.cc +++ b/tests/ConsumerTest.cc @@ -912,6 +912,12 @@ TEST(ConsumerTest, testReceiveAsyncAfterTopicTerminated) { 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(); From 23136c41f3d95f38928638869913519be246fcc8 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 12:17:18 -0700 Subject: [PATCH 10/12] st: queue consumer drain, robustness, and honesty fixes from #605 review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Ack loss on drained sealed segments: end-of-topic only means the classic prefetch queue drained — messages already fanned into the mux queue or held by the application still need the segment consumer to route their acks. Track outstanding (fanned-in minus acked/nacked) messages per segment and defer the drain-close until the count reaches zero; the consumer stays in the map for ack routing meanwhile. - Receive-loop recursion: receiveAsync completes inline when a message is prefetched and offer()'s future is already complete while the queue has room, so the re-arm chain grew the stack once per message. Hop the re-arm through the IO executor so the chain is a loop again. - A segment subscribe that failed off the first-layout path was only retried on the next DAG push, which may not come for hours: back it with a bounded backoff retry (10 attempts, 100->500ms, the producer's constants), skipping segments that left the DAG or drained. - Message::topic() reported the internal segment:// backing topic; pass the scalable topic as the override so the public contract holds. - A configured deadLetterPolicy was silently ignored; it now fails the subscribe with ResultOperationNotSupported, and the API docs say so, until dead-lettering lands. - ReceiveQueue timed receives never cancelled their timer when a message won the race, accumulating live timers proportional to receive rate x timeout; park {promise, timer} together and cancel on delivery and on close. --- include/pulsar/st/QueueConsumer.h | 6 +- lib/st/QueueConsumerImpl.cc | 164 +++++++++++++++++++++++++----- lib/st/QueueConsumerImpl.h | 40 +++++++- lib/st/ReceiveQueue.cc | 58 ++++++++--- lib/st/ReceiveQueue.h | 22 ++-- 5 files changed, 237 insertions(+), 53 deletions(-) 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/lib/st/QueueConsumerImpl.cc b/lib/st/QueueConsumerImpl.cc index 695b2964..62604b09 100644 --- a/lib/st/QueueConsumerImpl.cc +++ b/lib/st/QueueConsumerImpl.cc @@ -18,7 +18,10 @@ */ #include "QueueConsumerImpl.h" +#include +#include #include +#include #include #include #include @@ -40,6 +43,16 @@ pulsar::InitialPosition toClassicInitialPosition(SubscriptionInitialPosition pos : 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) @@ -48,11 +61,19 @@ QueueConsumerImpl::QueueConsumerImpl(pulsar::ClientImplPtr classic, QueueConsume topic_(config_.topic), subscription_(config_.subscriptionName), consumerName_(config_.consumerName.value_or(std::string{})), - receiveQueue_( - std::make_shared(classic_->getIOExecutorProvider()->get(), kReceiveQueueCapacity)), + 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( @@ -92,6 +113,8 @@ void QueueConsumerImpl::onLayoutChange(const SegmentLayout& newLayout, const Seg 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; @@ -109,14 +132,7 @@ void QueueConsumerImpl::onLayoutChange(const SegmentLayout& newLayout, const Seg } } - for (auto& future : retired) { - future.addListener([](const Expected& result) { - if (result) { - pulsar::Consumer consumer = *result; - consumer.closeAsync([](pulsar::Result) {}); - } - }); - } + for (auto& future : retired) closeWhenReady(future); if (first) { if (toAdd.empty()) { @@ -135,7 +151,10 @@ void QueueConsumerImpl::onLayoutChange(const SegmentLayout& newLayout, const Seg }); } } else { - for (const auto& segment : toAdd) getOrCreateSegmentConsumerAsync(segment); // best-effort + // 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); } } @@ -195,36 +214,122 @@ Future QueueConsumerImpl::getOrCreateSegmentConsumerAsync(cons 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) { - // A sealed segment fully drained: close its consumer, drop it from the cache, and - // remember it drained so a later reconcile does not re-subscribe the still-in-DAG - // sealed segment (which would redeliver its unacked messages as duplicates). + // 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_); - self->segmentConsumers_.erase(segmentId); - self->drainedSegments_.insert(segmentId); + auto outstanding = self->outstanding_.find(segmentId); + if (outstanding == self->outstanding_.end() || outstanding->second == 0) { + toClose = self->takeDrainedSegmentLocked(segmentId); + } else { + self->terminatedSegments_.insert(segmentId); + } } - pulsar::Consumer done = consumer; - done.closeAsync([](pulsar::Result) {}); + if (toClose) closeWhenReady(*toClose); } // Otherwise (AlreadyClosed / consumer closing) just stop the loop. return; } MessageId id = MessageIdFactory::create(message.getMessageId(), static_cast(segmentId)); - auto messageImpl = std::make_shared(message, std::move(id)); - // Re-arm only once the fan-in queue has room, so a slow consumer throttles this segment. + // 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->startReceiveLoop(consumer, segmentId); + 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) { @@ -249,11 +354,15 @@ void QueueConsumerImpl::acknowledge(const MessageId& id) { const auto& impl = MessageIdFactory::impl(id); if (!impl) return; const pulsar::MessageId v4 = impl->v4MessageId; - segmentConsumerFor(id).addListener([v4](const Expected& result) { + 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); }); } @@ -267,11 +376,17 @@ void QueueConsumerImpl::negativeAcknowledge(const MessageId& id) { const auto& impl = MessageIdFactory::impl(id); if (!impl) return; const pulsar::MessageId v4 = impl->v4MessageId; - segmentConsumerFor(id).addListener([v4](const Expected& result) { + 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); }); } @@ -290,6 +405,9 @@ Future QueueConsumerImpl::closeAsync() { consumers.reserve(segmentConsumers_.size()); for (auto& [segmentId, future] : segmentConsumers_) consumers.push_back(future); segmentConsumers_.clear(); + outstanding_.clear(); + terminatedSegments_.clear(); + drainedSegments_.clear(); } detail::Promise promise; diff --git a/lib/st/QueueConsumerImpl.h b/lib/st/QueueConsumerImpl.h index a79cbc8c..d4a7c448 100644 --- a/lib/st/QueueConsumerImpl.h +++ b/lib/st/QueueConsumerImpl.h @@ -24,9 +24,11 @@ #include #include +#include #include #include #include +#include #include #include #include @@ -36,6 +38,7 @@ #include "ReceiveQueue.h" #include "SegmentLayout.h" #include "lib/ClientImpl.h" +#include "lib/ExecutorService.h" namespace pulsar::st { @@ -50,7 +53,9 @@ namespace pulsar::st { * 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. + * 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: @@ -73,11 +78,28 @@ class QueueConsumerImpl : public std::enable_shared_from_this 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); @@ -92,6 +114,9 @@ class QueueConsumerImpl : public std::enable_shared_from_this 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_; @@ -101,9 +126,16 @@ class QueueConsumerImpl : public std::enable_shared_from_this bool sawFirstLayout_ = false; // guarded by mutex_ std::shared_ptr currentLayout_; // guarded by mutex_ std::unordered_map> segmentConsumers_; // guarded by mutex_ - // Segments that have reported end-of-topic and been drained; 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_. + // 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_; }; diff --git a/lib/st/ReceiveQueue.cc b/lib/st/ReceiveQueue.cc index 76290923..b8407a34 100644 --- a/lib/st/ReceiveQueue.cc +++ b/lib/st/ReceiveQueue.cc @@ -52,7 +52,7 @@ Future ReceiveQueue::receiveAsync() { buffer_.pop_front(); toSignal = takeCapacityWaitersIfRoomLocked(); } else { - pendingReceives_.emplace(nextReceiveId_++, promise); + pendingReceives_.emplace(nextReceiveId_++, PendingReceive{promise, nullptr}); } } for (auto& waiter : toSignal) waiter.setSuccess(); @@ -78,7 +78,7 @@ Future ReceiveQueue::receiveAsync(std::chrono::milliseconds time toSignal = takeCapacityWaitersIfRoomLocked(); } else { receiveId = nextReceiveId_++; - pendingReceives_.emplace(receiveId, promise); + pendingReceives_.emplace(receiveId, PendingReceive{promise, nullptr}); parked = true; } } @@ -88,25 +88,40 @@ Future ReceiveQueue::receiveAsync(std::chrono::milliseconds time 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(); - 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 - { - 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); + bool armed = false; + { + std::lock_guard lock(mutex_); + auto it = pendingReceives_.find(receiveId); + if (it != pendingReceives_.end()) { + it->second.timer = timer; + armed = true; } - promise.setError(Error{ResultTimeout, "receive timed out"}); - }); + } + // 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; @@ -117,7 +132,8 @@ Future ReceiveQueue::offer(MessageImplPtr message) { } else { if (!pendingReceives_.empty()) { auto oldest = pendingReceives_.begin(); // FIFO: lowest id - receiver = std::move(oldest->second); + receiver = std::move(oldest->second.promise); + receiverTimer = std::move(oldest->second.timer); pendingReceives_.erase(oldest); deliver = true; } else { @@ -130,6 +146,10 @@ Future ReceiveQueue::offer(MessageImplPtr message) { } } } + if (receiverTimer) { + ASIO_ERROR ignored; + receiverTimer->cancel(ignored); + } if (deliver) receiver.setValue(std::move(message)); if (hasRoom) { detail::Promise ready; @@ -140,7 +160,7 @@ Future ReceiveQueue::offer(MessageImplPtr message) { } void ReceiveQueue::close() { - std::map> pending; + std::map pending; std::deque> waiters; { std::lock_guard lock(mutex_); @@ -150,7 +170,13 @@ void ReceiveQueue::close() { waiters.swap(capacityWaiters_); buffer_.clear(); } - for (auto& [id, promise] : pending) promise.setError(Error{ResultAlreadyClosed, "consumer is closed"}); + 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 } diff --git a/lib/st/ReceiveQueue.h b/lib/st/ReceiveQueue.h index fe071d15..762ffa0a 100644 --- a/lib/st/ReceiveQueue.h +++ b/lib/st/ReceiveQueue.h @@ -28,10 +28,7 @@ #include #include -namespace pulsar { -class ExecutorService; -using ExecutorServicePtr = std::shared_ptr; -} // namespace pulsar +#include "lib/ExecutorService.h" namespace pulsar::st { @@ -65,15 +62,22 @@ class ReceiveQueue : public std::enable_shared_from_this { // 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_ + 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; From 60c763bafba1205f16c9621e3be29ffdc80ad92f Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 12:17:18 -0700 Subject: [PATCH 11/12] st: e2e for draining a sealed segment's backlog with sticking acks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scenario the queue consumer exists for — a split seals the parent WITHOUT migrating its backlog — had no coverage: both existing e2e tests split before producing, so the sealed parent was always empty and no end-of-topic was ever delivered. Produce 1200 messages (more than the classic prefetch queue and the mux capacity), split, then consume: every pre-split message must arrive through the sealed parent, and reattaching a second consumer on the same subscription must receive nothing — proving the acks routed through the drain-deferred close instead of being dropped. Fails on the pre-fix code. --- tests/st/StQueueConsumerE2ETest.cc | 96 ++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/tests/st/StQueueConsumerE2ETest.cc b/tests/st/StQueueConsumerE2ETest.cc index 67771df1..eef024fc 100644 --- a/tests/st/StQueueConsumerE2ETest.cc +++ b/tests/st/StQueueConsumerE2ETest.cc @@ -200,4 +200,100 @@ TEST(StQueueConsumerE2ETest, testConsumeAcrossSplitSegments) { 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 From 1cbbaa5c535f71dd39115c085abf1b2ebfc453df Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 14:38:43 -0700 Subject: [PATCH 12/12] Pin jidicula/clang-format-action to its commit SHA for the ASF actions policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since mid-August every new PR-validation run on this repo fails at workflow startup (startup_failure, 0s, "workflow file issue") with no workflow change on main — the same repo-wide pattern as the docker/build-push-action break fixed by #602. The ASF GitHub Actions policy requires external actions to be pinned to a specific git hash, and jidicula/clang-format-action@v4.11.0 was the one remaining tag-pinned external action after #602 pinned the docker ones. Pin it to the commit the v4.11.0 tag points to (f62da5e, unchanged behavior). --- .github/workflows/ci-pr-validation.yaml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci-pr-validation.yaml b/.github/workflows/ci-pr-validation.yaml index b1d83cc2..4ef96e75 100644 --- a/.github/workflows/ci-pr-validation.yaml +++ b/.github/workflows/ci-pr-validation.yaml @@ -35,7 +35,9 @@ jobs: steps: - uses: actions/checkout@v3 - name: Run clang-format style check for C/C++/Protobuf programs. - uses: jidicula/clang-format-action@v4.11.0 + # v4.11.0, pinned to its commit SHA per the ASF GitHub Actions policy (external actions + # must be pinned to a git hash; a tag reference fails workflow startup validation). + uses: jidicula/clang-format-action@f62da5e3d3a2d88ff364771d9d938773a618ab5e with: clang-format-version: '11' exclude-regex: '.*\.(proto|hpp)'