diff --git a/include/pulsar/st/detail/ProducerCore.h b/include/pulsar/st/detail/ProducerCore.h index 004dc2ee..13059a34 100644 --- a/include/pulsar/st/detail/ProducerCore.h +++ b/include/pulsar/st/detail/ProducerCore.h @@ -33,6 +33,7 @@ namespace pulsar::st { class ProducerImplBase; using ProducerImplPtr = std::shared_ptr; struct OutgoingMessage; +class ClientImpl; // lib/st — mints producer cores from createProducerAsync namespace detail { @@ -58,6 +59,7 @@ class PULSAR_PUBLIC ProducerCore { private: friend class ClientCore; + friend class ::pulsar::st::ClientImpl; explicit ProducerCore(ProducerImplPtr impl) : impl_(std::move(impl)) {} ProducerImplPtr impl_; diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc index dc79fce0..510456f4 100644 --- a/lib/ClientImpl.cc +++ b/lib/ClientImpl.cc @@ -94,6 +94,15 @@ std::string generateRandomName() { typedef std::vector StringList; +// segment:// topics are the internal backing topics of a scalable topic and are +// reachable only through the pulsar::st client, which bypasses this rejection. +static Error segmentTopicRejected(const std::string& topic) { + return Error{ResultInvalidTopicName, + "segment:// topics are the internal backing topics of a scalable topic; use the " + "pulsar::st API instead: " + + topic}; +} + static LookupServicePtr defaultLookupServiceFactory(const ServiceInfo& serviceInfo, const ClientConfiguration& clientConfiguration, ConnectionPool& pool) { @@ -204,6 +213,21 @@ LookupServicePtr ClientImpl::getLookup(const std::string& redirectedClusterURI) void ClientImpl::createProducerAsync(const std::string& topic, const ProducerConfiguration& conf, CreateProducerV2Callback callback, bool autoDownloadSchema) { + createProducerAsyncImpl(topic, conf, std::move(callback), autoDownloadSchema, + /* allowSegmentTopic */ false); +} + +void ClientImpl::createSegmentProducerAsync(const std::string& topic, const ProducerConfiguration& conf, + CreateProducerV2Callback callback, + const std::optional& assignedBrokerUrl) { + createProducerAsyncImpl(topic, conf, std::move(callback), /* autoDownloadSchema */ false, + /* allowSegmentTopic */ true, assignedBrokerUrl); +} + +void ClientImpl::createProducerAsyncImpl(const std::string& topic, const ProducerConfiguration& conf, + CreateProducerV2Callback callback, bool autoDownloadSchema, + bool allowSegmentTopic, + const std::optional& assignedBrokerUrl) { if (conf.isChunkingEnabled() && conf.getBatchingEnabled()) { throw std::invalid_argument("Batching and chunking of messages can't be enabled together"); } @@ -222,32 +246,38 @@ void ClientImpl::createProducerAsync(const std::string& topic, const ProducerCon return; } } + if (topicName->isSegment() && !allowSegmentTopic) { + callback(segmentTopicRejected(topic)); + return; + } if (autoDownloadSchema) { - getSchema(topicName).addListener([self{shared_from_this()}, topicName, callback{std::move(callback)}]( - const Error& error, const SchemaInfo& topicSchema) mutable { - if (error.result != ResultOk) { - callback(error); - return; - } - ProducerConfiguration conf; - conf.setSchema(topicSchema); - self->getPartitionMetadataAsync(topicName).addListener( - std::bind(&ClientImpl::handleCreateProducer, self, std::placeholders::_1, - std::placeholders::_2, topicName, conf, callback)); - }); + getSchema(topicName).addListener( + [self{shared_from_this()}, topicName, callback{std::move(callback)}, assignedBrokerUrl]( + const Error& error, const SchemaInfo& topicSchema) mutable { + if (error.result != ResultOk) { + callback(error); + return; + } + ProducerConfiguration conf; + conf.setSchema(topicSchema); + self->getPartitionMetadataAsync(topicName).addListener( + std::bind(&ClientImpl::handleCreateProducer, self, std::placeholders::_1, + std::placeholders::_2, topicName, conf, callback, assignedBrokerUrl)); + }); } else { getPartitionMetadataAsync(topicName).addListener( - [this, conf, topicName, callback{std::move(callback)}]( + [this, conf, topicName, callback{std::move(callback)}, assignedBrokerUrl]( const Error& error, const LookupDataResultPtr& partitionMetadata) { - handleCreateProducer(error, partitionMetadata, topicName, conf, callback); + handleCreateProducer(error, partitionMetadata, topicName, conf, callback, assignedBrokerUrl); }); } } void ClientImpl::handleCreateProducer(const Error& error, const LookupDataResultPtr& partitionMetadata, const TopicNamePtr& topicName, const ProducerConfiguration& conf, - CreateProducerV2Callback callback) { + CreateProducerV2Callback callback, + const std::optional& assignedBrokerUrl) { if (!error.result) { ProducerImplBasePtr producer; @@ -258,7 +288,10 @@ void ClientImpl::handleCreateProducer(const Error& error, const LookupDataResult producer = std::make_shared( shared_from_this(), topicName, partitionMetadata->getPartitions(), conf, interceptors); } else { - producer = std::make_shared(shared_from_this(), *topicName, conf, interceptors); + producer = + std::make_shared(shared_from_this(), *topicName, conf, interceptors, + /* partition */ -1, + /* retryOnCreationError */ false, assignedBrokerUrl); } } catch (const std::runtime_error& e) { LOG_ERROR("Failed to create producer: " << e.what()); @@ -319,6 +352,10 @@ void ClientImpl::createReaderAsyncV2(const std::string& topic, const MessageId& return; } } + if (topicName->isSegment()) { + callback(segmentTopicRejected(topic)); + return; + } getPartitionMetadataAsync(topicName).addListener( [this, self{shared_from_this()}, topicName, startMessageId, conf, callback{std::move(callback)}]( @@ -348,6 +385,10 @@ void ClientImpl::createTableViewAsyncV2(const std::string& topic, const TableVie return; } } + if (topicName->isSegment()) { + callback(segmentTopicRejected(topic)); + return; + } TableViewImplPtr tableViewPtr = std::make_shared(shared_from_this(), topicName->toString(), conf); @@ -586,6 +627,11 @@ void ClientImpl::subscribeToTopicsAsyncV2(const std::string& topic, const std::s } } + if (topicName->isSegment()) { + callback(segmentTopicRejected(topic)); + return; + } + getPartitionMetadataAsync(topicName).addListener( [this, self{shared_from_this()}, topicName, subscriptionName, conf, callback{std::move(callback)}]( const auto& error, const auto& metadata) { @@ -768,6 +814,11 @@ void ClientImpl::getPartitionsForTopicAsync(const std::string& topic, const GetP return; } } + if (topicName->isSegment()) { + LOG_ERROR(segmentTopicRejected(topic)); + callback(ResultInvalidTopicName, StringList()); + return; + } getPartitionMetadataAsync(topicName).addListener( [this, self{shared_from_this()}, topicName, callback](const auto& error, const auto& metadata) { handleGetPartitions(error.result, metadata, topicName, callback); diff --git a/lib/ClientImpl.h b/lib/ClientImpl.h index cdabf1fc..7b822c08 100644 --- a/lib/ClientImpl.h +++ b/lib/ClientImpl.h @@ -95,6 +95,14 @@ class ClientImpl : public std::enable_shared_from_this { void createProducerAsync(const std::string& topic, const ProducerConfiguration& conf, CreateProducerV2Callback callback, bool autoDownloadSchema = false); + /** + * Scalable topics (pulsar::st): create a producer on a segment:// backing + * topic, bypassing the segment-domain rejection applied to the public path. + */ + void createSegmentProducerAsync(const std::string& topic, const ProducerConfiguration& conf, + CreateProducerV2Callback callback, + const std::optional& assignedBrokerUrl = std::nullopt); + void subscribeAsync(const std::string& topic, const std::string& subscriptionName, const ConsumerConfiguration& conf, const SubscribeCallback& callback); @@ -180,9 +188,15 @@ class ClientImpl : public std::enable_shared_from_this { friend class PulsarFriend; private: + void createProducerAsyncImpl(const std::string& topic, const ProducerConfiguration& conf, + CreateProducerV2Callback callback, bool autoDownloadSchema, + bool allowSegmentTopic, + const std::optional& assignedBrokerUrl = std::nullopt); + void handleCreateProducer(const Error& error, const LookupDataResultPtr& partitionMetadata, const TopicNamePtr& topicName, const ProducerConfiguration& conf, - CreateProducerV2Callback callback); + CreateProducerV2Callback callback, + const std::optional& assignedBrokerUrl = std::nullopt); void handleSubscribe(const Error& error, const LookupDataResultPtr& partitionMetadata, const TopicNamePtr& topicName, const std::string& consumerName, diff --git a/lib/HandlerBase.cc b/lib/HandlerBase.cc index 5bccc0bc..45b6a96e 100644 --- a/lib/HandlerBase.cc +++ b/lib/HandlerBase.cc @@ -54,11 +54,13 @@ HandlerBase::~HandlerBase() { cancelTimer(*creationTimer_); } -void HandlerBase::start() { +void HandlerBase::start() { start(std::nullopt); } + +void HandlerBase::start(const optional& assignedBrokerUrl) { // guard against concurrent state changes such as closing State state = NotStarted; if (state_.compare_exchange_strong(state, Pending)) { - grabCnx(); + grabCnx(assignedBrokerUrl); } creationTimer_->expires_after(operationTimeut_); auto weakSelf = weak_from_this(); diff --git a/lib/HandlerBase.h b/lib/HandlerBase.h index 16124156..811d2361 100644 --- a/lib/HandlerBase.h +++ b/lib/HandlerBase.h @@ -51,6 +51,13 @@ class HandlerBase : public std::enable_shared_from_this { void start(); + /* + * Start and, on the initial connect, go straight to @p assignedBrokerUrl instead of + * doing a topic lookup. Used by scalable-topics per-segment producers pinned to the + * DAG-provided owner broker. An empty optional behaves exactly like start(). + */ + void start(const optional& assignedBrokerUrl); + ClientConnectionWeakPtr getCnx() const; void setCnx(const ClientConnectionPtr& cnx); void resetCnx() { setCnx(nullptr); } diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc index 4697e72b..f1da6f59 100644 --- a/lib/ProducerImpl.cc +++ b/lib/ProducerImpl.cc @@ -50,7 +50,8 @@ using std::chrono::milliseconds; ProducerImpl::ProducerImpl(const ClientImplPtr& client, const TopicName& topicName, const ProducerConfiguration& conf, const ProducerInterceptorsPtr& interceptors, - int32_t partition, bool retryOnCreationError) + int32_t partition, bool retryOnCreationError, + const optional& assignedBrokerUrl) : HandlerBase(client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition), Backoff(milliseconds(client->getClientConfig().getInitialBackoffIntervalMs()), milliseconds(client->getClientConfig().getMaxBackoffIntervalMs()), @@ -58,6 +59,7 @@ ProducerImpl::ProducerImpl(const ClientImplPtr& client, const TopicName& topicNa conf_(conf), semaphore_(), partition_(partition), + assignedBrokerUrl_(assignedBrokerUrl), producerName_(conf_.getProducerName()), userProvidedProducerName_(false), producerStr_("[" + topic() + ", " + producerName_ + "] "), @@ -1007,7 +1009,7 @@ void ProducerImpl::disconnectProducer(const optional& assignedBroke void ProducerImpl::disconnectProducer() { disconnectProducer(std::nullopt); } void ProducerImpl::start() { - HandlerBase::start(); + HandlerBase::start(assignedBrokerUrl_); if (conf_.getLazyStartPartitionedProducers() && conf_.getAccessMode() == ProducerConfiguration::Shared) { // we need to kick it off now as it is possible that the connection may take diff --git a/lib/ProducerImpl.h b/lib/ProducerImpl.h index e7502533..6942fcd3 100644 --- a/lib/ProducerImpl.h +++ b/lib/ProducerImpl.h @@ -73,7 +73,8 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase { ProducerImpl(const ClientImplPtr& client, const TopicName& topic, const ProducerConfiguration& producerConfiguration, const ProducerInterceptorsPtr& interceptors, int32_t partition = -1, - bool retryOnCreationError = false); + bool retryOnCreationError = false, + const optional& assignedBrokerUrl = std::nullopt); ~ProducerImpl(); // overrided methods from ProducerImplBase @@ -177,6 +178,9 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase { std::list> pendingMessagesQueue_; const int32_t partition_; // -1 if topic is non-partitioned + // When set, the initial connect goes straight to this broker (no lookup) — used by + // scalable-topics per-segment producers pinned to the DAG-provided owner broker. + const optional assignedBrokerUrl_; std::string producerName_; bool userProvidedProducerName_; std::string producerStr_; diff --git a/lib/TopicName.cc b/lib/TopicName.cc index 487eee53..aed56641 100644 --- a/lib/TopicName.cc +++ b/lib/TopicName.cc @@ -35,6 +35,7 @@ namespace pulsar { const std::string TopicDomain::Persistent = "persistent"; const std::string TopicDomain::NonPersistent = "non-persistent"; +const std::string TopicDomain::Segment = "segment"; static const std::string PARTITION_NAME_SUFFIX = "-partition-"; typedef std::unique_lock Lock; @@ -105,7 +106,23 @@ bool TopicName::parse(const std::string& topicName, std::string& domain, std::st domain = pathTokens[0]; size_t numSlashIndexes; bool isV2Topic; - if (pathTokens.size() == 4) { + if (domain == TopicDomain::Segment) { + // segment://///: always the new + // (cluster-less) format, with a '/' inside the local name — so the + // token-count heuristic below must not mistake it for a legacy V1 name. + if (pathTokens.size() < 5) { + LOG_ERROR( + "Segment topic name is not valid, expected " + "segment:///// - " + << topicName); + return false; + } + property = pathTokens[1]; + cluster = ""; + namespacePortion = pathTokens[2]; + numSlashIndexes = 3; + isV2Topic = true; + } else if (pathTokens.size() == 4) { // New topic name without cluster name property = pathTokens[1]; cluster = ""; @@ -171,7 +188,12 @@ bool TopicName::operator==(const TopicName& other) const { bool TopicName::validate() { // Check if domain matches with TopicDomain::Persistent, in future check "memory" when server is // ready. - if (domain_.compare(TopicDomain::Persistent) != 0 && domain_.compare(TopicDomain::NonPersistent) != 0) { + if (domain_.compare(TopicDomain::Persistent) != 0 && domain_.compare(TopicDomain::NonPersistent) != 0 && + domain_.compare(TopicDomain::Segment) != 0) { + return false; + } + if (domain_ == TopicDomain::Segment && !isV2Topic_) { + // Segment topics only exist in the new (cluster-less) format. return false; } // cluster_ can be empty @@ -228,7 +250,13 @@ std::string TopicName::toString() const { return ss.str(); } -bool TopicName::isPersistent() const { return this->domain_ == TopicDomain::Persistent; } +// A segment topic is the persistent backing topic of one scalable-topic segment, +// so it counts as persistent (matching the Java TopicName). +bool TopicName::isPersistent() const { + return this->domain_ == TopicDomain::Persistent || this->domain_ == TopicDomain::Segment; +} + +bool TopicName::isSegment() const { return this->domain_ == TopicDomain::Segment; } std::string TopicName::getTopicPartitionName(unsigned int partition) const { std::stringstream topicPartitionName; diff --git a/lib/TopicName.h b/lib/TopicName.h index bee8138a..34fa55ab 100644 --- a/lib/TopicName.h +++ b/lib/TopicName.h @@ -37,6 +37,11 @@ class PULSAR_PUBLIC TopicDomain { public: static const std::string Persistent; static const std::string NonPersistent; + // The per-segment backing topics of a scalable topic (PIP-460): + // segment://///--. + // Internal to the pulsar::st client; the classic user-facing entry points + // reject this domain. + static const std::string Segment; }; // class TopicDomain class PULSAR_PUBLIC TopicName : public ServiceUnitId { @@ -62,6 +67,7 @@ class PULSAR_PUBLIC TopicName : public ServiceUnitId { std::string getEncodedLocalName() const; std::string toString() const; bool isPersistent() const; + bool isSegment() const; NamespaceNamePtr getNamespaceName(); int getPartitionIndex() const noexcept { return partition_; } static std::shared_ptr get(const std::string& topicName); diff --git a/lib/st/MessageIdImpl.h b/lib/st/MessageIdImpl.h new file mode 100644 index 00000000..0b3f9daf --- /dev/null +++ b/lib/st/MessageIdImpl.h @@ -0,0 +1,76 @@ +/** + * 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::st { + +/** + * The hidden state of a pulsar::st::MessageId, mirroring the Java client's + * MessageIdV5 field-for-field so the two clients' serialized ids interoperate: + * the classic per-segment message id plus the segment it belongs to, and the + * optional position-vector sections used by the consumer side (a cumulative-ack + * StreamConsumer id also carries the other segments' positions; a namespace + * consumer's id carries them per topic). The producer path fills only + * v4MessageId + segmentId. + */ +class MessageIdImpl { + public: + /** segmentId of the earliest/latest sentinels, which span all segments. */ + static constexpr std::int64_t kNoSegment = -1; + + pulsar::MessageId v4MessageId; + std::int64_t segmentId = kNoSegment; + /** Per-segment positions for cumulative-ack ids. Empty on the producer path. */ + std::map positionVector; + /** The scalable topic this id belongs to, when carried (topic://...). */ + std::optional parentTopic; + /** Cross-topic position vectors for namespace consumers, when carried. */ + std::optional>> multiTopicVector; +}; + +/** + * INTERNAL factory befriended by the public MessageId handle: the only way + * lib/st mints typed ids and reaches back into their hidden state. + */ +class MessageIdFactory { + public: + static MessageId create(std::shared_ptr impl) { return MessageId(std::move(impl)); } + + /** The producer path: an id for one published message in one segment. */ + static MessageId create(const pulsar::MessageId& v4MessageId, std::int64_t segmentId) { + auto impl = std::make_shared(); + impl->v4MessageId = v4MessageId; + impl->segmentId = segmentId; + return MessageId(std::move(impl)); + } + + static const std::shared_ptr& impl(const MessageId& id) { return id.impl_; } +}; + +} // namespace pulsar::st diff --git a/lib/st/ProducerCore.cc b/lib/st/ProducerCore.cc new file mode 100644 index 00000000..6629b7cf --- /dev/null +++ b/lib/st/ProducerCore.cc @@ -0,0 +1,39 @@ +/** + * 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 // complete OutgoingMessage (moved into the impl) +#include + +#include + +#include "ProducerImplBase.h" + +namespace pulsar::st::detail { + +// Thin forwarders to the hidden ProducerImplBase. A const ProducerCore still reaches a +// non-const pointee through the shared_ptr, so the mutating impl methods need no const. +Future ProducerCore::sendAsync(OutgoingMessage message) const { + return impl_->sendAsync(std::move(message)); +} +std::string_view ProducerCore::topic() const { return impl_->topic(); } +std::string_view ProducerCore::producerName() const { return impl_->producerName(); } +std::optional ProducerCore::lastSequenceId() const { return impl_->lastSequenceId(); } +Future ProducerCore::flushAsync() const { return impl_->flushAsync(); } +Future ProducerCore::closeAsync() const { return impl_->closeAsync(); } + +} // namespace pulsar::st::detail diff --git a/lib/st/ProducerImplBase.h b/lib/st/ProducerImplBase.h new file mode 100644 index 00000000..156b2e91 --- /dev/null +++ b/lib/st/ProducerImplBase.h @@ -0,0 +1,58 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#pragma once + +#include +#include + +#include +#include +#include + +namespace pulsar::st { + +struct OutgoingMessage; + +/** + * INTERNAL — the abstract producer the public `detail::ProducerCore` forwards to. + * + * `detail::ProducerCore` holds a `std::shared_ptr` and each of its + * methods is a one-line forward to the matching virtual here; `StProducerImpl` (the + * scalable-topics producer) is the concrete implementation. This lives in `pulsar::st` + * (not `detail`) to match the forward declaration in `detail/ProducerCore.h`. It is a + * different type from the classic `pulsar::ProducerImplBase` — different namespace, no + * collision. + * + * The observer methods are `const`; the mutating ones are not. `ProducerCore`'s methods + * are all `const`, but a `const` handle still reaches a non-const pointee through the + * shared_ptr, so the mutating virtuals need not be const. + */ +class ProducerImplBase { + public: + virtual ~ProducerImplBase() = default; + + virtual Future sendAsync(OutgoingMessage message) = 0; + virtual std::string_view topic() const = 0; + virtual std::string_view producerName() const = 0; + virtual std::optional lastSequenceId() const = 0; + virtual Future flushAsync() = 0; + virtual Future closeAsync() = 0; +}; + +} // namespace pulsar::st diff --git a/lib/st/StClientImpl.cc b/lib/st/StClientImpl.cc index 8b38ab9a..40ec5b7e 100644 --- a/lib/st/StClientImpl.cc +++ b/lib/st/StClientImpl.cc @@ -21,6 +21,8 @@ #include #include +#include "StProducerImpl.h" + namespace pulsar::st { namespace { @@ -41,14 +43,24 @@ Future notImplementedYet(const char* what) { ClientImpl::ClientImpl(pulsar::ClientImplPtr classicClient, const TransactionPolicy& transactionPolicy) : classic_(std::move(classicClient)), transactionPolicy_(transactionPolicy) {} -// The producer/consumer/transaction paths land in follow-up phases. Each takes -// its config by value (the sink the real implementation will move from), but as -// a stub it does not consume the config yet — hence the value-param suppressions. -// NOLINTNEXTLINE(performance-unnecessary-value-param) -Future ClientImpl::createProducerAsync(ProducerConfig) { - return notImplementedYet("createProducer"); +Future ClientImpl::createProducerAsync(ProducerConfig 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::ProducerCore{impl}); + } else { + promise.setError(result.error()); + } + }); + return promise.getFuture(); } +// The consumer/transaction paths land in follow-up phases. Each takes its config by +// value (the sink the real implementation will move from), but as a stub it does not +// consume the config yet — hence the value-param suppressions. + // NOLINTNEXTLINE(performance-unnecessary-value-param) Future ClientImpl::subscribeStreamAsync(StreamConsumerConfig) { return notImplementedYet("subscribeStream"); diff --git a/lib/st/StMessageId.cc b/lib/st/StMessageId.cc index 4bfa037a..893dcb01 100644 --- a/lib/st/StMessageId.cc +++ b/lib/st/StMessageId.cc @@ -18,10 +18,273 @@ */ #include +#include +#include +#include +#include +#include +#include + +#include "MessageIdImpl.h" +#include "lib/LogUtils.h" + +DECLARE_LOG_OBJECT() + namespace pulsar::st { +namespace { + +// The serialized form is the Java client's MessageIdV5 layout (big-endian): +// int64 segmentId +// int32 v4Length, v4 bytes (classic MessageId serialization) +// int32 positionCount, {int64 segmentId, int32 length, bytes} entries +// int32 parentTopicLength (-1 = absent), utf-8 bytes +// int32 multiTopicCount (-1 = absent), +// {int32 topicLength, bytes, int32 positionCount, entries} entries +// The trailing sections are optional on read, so shorter (older) forms parse. + +void putInt32(std::vector& out, std::int32_t value) { + auto u = static_cast(value); + for (int shift = 24; shift >= 0; shift -= 8) { + out.push_back(static_cast((u >> shift) & 0xFF)); + } +} + +void putInt64(std::vector& out, std::int64_t value) { + auto u = static_cast(value); + for (int shift = 56; shift >= 0; shift -= 8) { + out.push_back(static_cast((u >> shift) & 0xFF)); + } +} + +void putBytes(std::vector& out, const std::string& bytes) { + const auto* p = reinterpret_cast(bytes.data()); + out.insert(out.end(), p, p + bytes.size()); +} + +class Reader { + public: + explicit Reader(std::span data) : data_(data) {} + + bool hasRemaining() const { return pos_ < data_.size(); } + + bool readInt32(std::int32_t& value) { + if (data_.size() - pos_ < 4) return false; + std::uint32_t u = 0; + for (int i = 0; i < 4; i++) { + u = (u << 8) | std::to_integer(data_[pos_++]); + } + value = static_cast(u); + return true; + } + + bool readInt64(std::int64_t& value) { + if (data_.size() - pos_ < 8) return false; + std::uint64_t u = 0; + for (int i = 0; i < 8; i++) { + u = (u << 8) | std::to_integer(data_[pos_++]); + } + value = static_cast(u); + return true; + } + + bool readBytes(std::int32_t length, std::string& out) { + if (length < 0 || static_cast(length) > data_.size() - pos_) return false; + out.assign(reinterpret_cast(data_.data() + pos_), static_cast(length)); + pos_ += static_cast(length); + return true; + } + + private: + std::span data_; + std::size_t pos_ = 0; +}; + +void writePositionVector(std::vector& out, + const std::map& vector) { + putInt32(out, static_cast(vector.size())); + for (const auto& entry : vector) { + putInt64(out, entry.first); + std::string serialized; + entry.second.serialize(serialized); + putInt32(out, static_cast(serialized.size())); + putBytes(out, serialized); + } +} + +bool readPositionVector(Reader& reader, std::map& out) { + std::int32_t count; + if (!reader.readInt32(count) || count < 0) return false; + for (std::int32_t i = 0; i < count; i++) { + std::int64_t segmentId; + std::int32_t length; + std::string bytes; + if (!reader.readInt64(segmentId) || !reader.readInt32(length) || !reader.readBytes(length, bytes)) { + return false; + } + out.emplace(segmentId, pulsar::MessageId::deserialize(bytes)); + } + return true; +} + +} // namespace + // An empty id: impl_ stays null, so the handle is falsy under operator bool and // compares equal only to other empty ids. MessageId::MessageId() = default; +MessageId::MessageId(std::shared_ptr impl) : impl_(std::move(impl)) {} + +const MessageId& MessageId::earliest() { + static const MessageId id = + MessageIdFactory::create(pulsar::MessageId::earliest(), MessageIdImpl::kNoSegment); + return id; +} + +const MessageId& MessageId::latest() { + static const MessageId id = + MessageIdFactory::create(pulsar::MessageId::latest(), MessageIdImpl::kNoSegment); + return id; +} + +std::vector MessageId::toByteArray() const { + if (!impl_) { + return {}; + } + std::vector out; + putInt64(out, impl_->segmentId); + + std::string v4Bytes; + impl_->v4MessageId.serialize(v4Bytes); + putInt32(out, static_cast(v4Bytes.size())); + putBytes(out, v4Bytes); + + writePositionVector(out, impl_->positionVector); + + if (impl_->parentTopic) { + putInt32(out, static_cast(impl_->parentTopic->size())); + putBytes(out, *impl_->parentTopic); + } else { + putInt32(out, -1); + } + + if (impl_->multiTopicVector) { + putInt32(out, static_cast(impl_->multiTopicVector->size())); + for (const auto& entry : *impl_->multiTopicVector) { + putInt32(out, static_cast(entry.first.size())); + putBytes(out, entry.first); + writePositionVector(out, entry.second); + } + } else { + putInt32(out, -1); + } + return out; +} + +MessageId MessageId::fromByteArray(std::span data) { + try { + Reader reader(data); + auto impl = std::make_shared(); + + std::int32_t v4Length; + std::string v4Bytes; + if (!reader.readInt64(impl->segmentId) || !reader.readInt32(v4Length) || + !reader.readBytes(v4Length, v4Bytes)) { + LOG_WARN("Discarding malformed scalable-topic MessageId: " << data.size() << " bytes"); + return MessageId(); + } + impl->v4MessageId = pulsar::MessageId::deserialize(v4Bytes); + + if (reader.hasRemaining() && !readPositionVector(reader, impl->positionVector)) { + LOG_WARN("Discarding scalable-topic MessageId with a malformed position vector"); + return MessageId(); + } + + if (reader.hasRemaining()) { + std::int32_t parentLength; + if (!reader.readInt32(parentLength)) { + return MessageId(); + } + if (parentLength >= 0) { + std::string parent; + if (!reader.readBytes(parentLength, parent)) { + return MessageId(); + } + impl->parentTopic = std::move(parent); + } + } + + if (reader.hasRemaining()) { + std::int32_t topicCount; + if (!reader.readInt32(topicCount)) { + return MessageId(); + } + if (topicCount >= 0) { + std::map> multi; + for (std::int32_t i = 0; i < topicCount; i++) { + std::int32_t topicLength; + std::string topic; + std::map inner; + if (!reader.readInt32(topicLength) || !reader.readBytes(topicLength, topic) || + !readPositionVector(reader, inner)) { + return MessageId(); + } + multi.emplace(std::move(topic), std::move(inner)); + } + impl->multiTopicVector = std::move(multi); + } + } + return MessageIdFactory::create(std::move(impl)); + } catch (const std::exception& e) { + // The classic MessageId deserialization throws on malformed proto bytes. + LOG_WARN("Discarding malformed scalable-topic MessageId: " << e.what()); + return MessageId(); + } +} + +std::strong_ordering MessageId::operator<=>(const MessageId& other) const { + // Empty ids sort before every real id and tie with each other. + if (!impl_ || !other.impl_) { + return static_cast(impl_) <=> static_cast(other.impl_); + } + if (auto cmp = impl_->segmentId <=> other.impl_->segmentId; cmp != 0) { + return cmp; + } + if (impl_->v4MessageId == other.impl_->v4MessageId) { + return std::strong_ordering::equal; + } + return impl_->v4MessageId < other.impl_->v4MessageId ? std::strong_ordering::less + : std::strong_ordering::greater; +} + +bool MessageId::operator==(const MessageId& other) const { + if (!impl_ || !other.impl_) { + return static_cast(impl_) == static_cast(other.impl_); + } + return impl_->segmentId == other.impl_->segmentId && impl_->v4MessageId == other.impl_->v4MessageId; +} + +std::ostream& operator<<(std::ostream& s, const MessageId& messageId) { + if (!messageId.impl_) { + return s << "{empty}"; + } + return s << "{segment=" << messageId.impl_->segmentId << ", id=" << messageId.impl_->v4MessageId + << ", positions=" << messageId.impl_->positionVector.size() << "}"; +} + } // namespace pulsar::st + +std::size_t std::hash::operator()( + const pulsar::st::MessageId& messageId) const noexcept { + const auto& impl = pulsar::st::MessageIdFactory::impl(messageId); + if (!impl) { + return 0; + } + std::size_t seed = std::hash()(impl->segmentId); + auto combine = [&seed](std::size_t h) { seed ^= h + 0x9e3779b9 + (seed << 6) + (seed >> 2); }; + combine(std::hash()(impl->v4MessageId.ledgerId())); + combine(std::hash()(impl->v4MessageId.entryId())); + combine(std::hash()(impl->v4MessageId.batchIndex())); + combine(std::hash()(impl->v4MessageId.partition())); + return seed; +} diff --git a/lib/st/StProducerImpl.cc b/lib/st/StProducerImpl.cc new file mode 100644 index 00000000..aedd2242 --- /dev/null +++ b/lib/st/StProducerImpl.cc @@ -0,0 +1,460 @@ +/** + * 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 "StProducerImpl.h" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "MessageIdImpl.h" +#include "lib/ExecutorService.h" +#include "lib/LogUtils.h" + +DECLARE_LOG_OBJECT() + +namespace pulsar::st { + +StProducerImpl::StProducerImpl(pulsar::ClientImplPtr classic, ProducerConfig config) + : classic_(std::move(classic)), + config_(std::move(config)), + topic_(config_.topic), + producerName_(config_.producerName.value_or(std::string{})), + currentLayout_(std::make_shared()) {} + +Future StProducerImpl::start() { + dagWatch_ = std::make_shared(classic_, config_.topic, /*createIfMissing*/ true); + std::weak_ptr weak = weak_from_this(); + // Register the layout listener BEFORE start() so the very first layout is delivered to it. + 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 StProducerImpl::onStartResult(const Expected& result) { + // Only the failure path is handled here: the DagWatchSession fails start()'s future (without + // invoking the layout listener) when the lookup fails before the first layout arrives. On + // success the listener runs immediately after and drives completeStart(). + if (!result) { + startPromise_.setError(result.error()); + } +} + +void StProducerImpl::onLayoutChange(const SegmentLayout& newLayout, const SegmentLayout& /*oldLayout*/) { + std::vector> retired; + bool first = false; + { + std::lock_guard lock(mutex_); + first = !sawFirstLayout_; + sawFirstLayout_ = true; + currentLayout_ = std::make_shared(newLayout); + + std::unordered_set active; + for (const auto& segment : newLayout.activeSegments()) active.insert(segment.segmentId); + for (auto it = segmentProducers_.begin(); it != segmentProducers_.end();) { + if (active.find(it->first) == active.end()) { + retired.push_back(std::move(it->second)); + it = segmentProducers_.erase(it); + } else { + ++it; + } + } + } + + // Close producers for segments that left the layout, fire-and-forget. + for (auto& future : retired) { + future.addListener([](const Expected& result) { + if (result) { + pulsar::Producer producer = *result; + producer.closeAsync([](pulsar::Result) {}); + } + }); + } + + if (first) { + completeStart(); + } else if (requiresExclusiveAttach(config_.accessMode)) { + // Exclusive modes claim newly-active segments eagerly (best-effort; failures are logged). + for (const auto& segment : newLayout.activeSegments()) { + getOrCreateSegmentProducerAsync(segment.segmentId); + } + } +} + +void StProducerImpl::completeStart() { + if (!requiresExclusiveAttach(config_.accessMode)) { + startPromise_.setSuccess(); // Shared: segment producers are created lazily on first send. + return; + } + auto layout = snapshotLayout(); + const auto& segments = layout->activeSegments(); + if (segments.empty()) { + startPromise_.setSuccess(); + return; + } + // Exclusive: eagerly attach every active segment; the create fails up front if the exclusive + // claim is refused, exactly like the Java strict eager attach. + auto remaining = std::make_shared>(static_cast(segments.size())); + for (const auto& segment : segments) { + getOrCreateSegmentProducerAsync(segment.segmentId) + .addListener([self = shared_from_this(), remaining](const Expected& result) { + if (!result) { + self->startPromise_.setError( + result.error()); // first error wins (complete is idempotent) + return; + } + if (remaining->fetch_sub(1) == 1) self->startPromise_.setSuccess(); + }); + } +} + +pulsar::ProducerConfiguration StProducerImpl::buildSegmentConfiguration(const Segment& segment) const { + // Build a FRESH config every time: pulsar::ProducerConfiguration's copy constructor shares its + // impl, so copy-then-mutate would clobber every other segment's config. + pulsar::ProducerConfiguration conf; + conf.setSchema(config_.schema); + conf.setAccessMode(toClassicAccessMode(config_.accessMode)); + if (config_.sendTimeoutMs) conf.setSendTimeout(static_cast(*config_.sendTimeoutMs)); + conf.setBlockIfQueueFull(config_.blockIfQueueFull); + if (config_.initialSequenceId) conf.setInitialSequenceId(*config_.initialSequenceId); + for (const auto& [key, value] : config_.properties) conf.setProperty(key, value); + if (config_.producerName) { + conf.setProducerName(*config_.producerName + "-seg-" + std::to_string(segment.segmentId)); + } + // Legacy segments wrap an externally-managed persistent:// topic; mark the connection so the + // regular->scalable migration precheck (PIP-475) recognizes it. Real segment:// topics aren't. + if (segment.isLegacy()) { + conf.setProperty("__pulsar.v5.managed", "true"); + } + // PIP-486 entry-bucketing: only the encryption-disables-batching branch is portable today; the + // EntryBucketBatcherBuilder branch has no C++ seam yet (TODO when a batcher-builder lands). + if (conf.isEncryptionEnabled()) { + conf.setBatchingEnabled(false); + } + return conf; +} + +Future StProducerImpl::getOrCreateSegmentProducerAsync(std::uint64_t segmentId) { + std::lock_guard lock(mutex_); + if (auto it = segmentProducers_.find(segmentId); it != segmentProducers_.end()) { + return it->second; // concurrent cold senders share the one creation attempt + } + + const Segment* segment = nullptr; + for (const auto& candidate : currentLayout_->activeSegments()) { + if (candidate.segmentId == segmentId) { + segment = &candidate; + break; + } + } + detail::Promise promise; + if (segment == nullptr) { + // Transient — don't cache, so a later routing attempt re-resolves against a fresh layout. + promise.setError(Error{ResultUnknownError, + "segment " + std::to_string(segmentId) + " is not in the active layout"}); + return promise.getFuture(); + } + + pulsar::ProducerConfiguration conf = buildSegmentConfiguration(*segment); + const std::string attachTopic = segment->attachTopicName(); + std::optional assignedBrokerUrl; + if (const std::string* url = currentLayout_->brokerUrl(segmentId)) { + assignedBrokerUrl = *url; // pin to the DAG-provided owner broker + } + + auto future = promise.getFuture(); + segmentProducers_.insert_or_assign(segmentId, future); + classic_->createSegmentProducerAsync( + attachTopic, conf, + [promise](std::variant result) { + if (auto* producer = std::get_if(&result)) { + promise.setValue(std::move(*producer)); + } else { + promise.setError(std::get(result)); + } + }, + assignedBrokerUrl); + return future; +} + +Future StProducerImpl::sendAsync(OutgoingMessage message) { + detail::Promise userPromise; + auto userFuture = userPromise.getFuture(); + + if (closed_.load()) { + userPromise.setError(Error{ResultAlreadyClosed, "producer is closed"}); + return userFuture; + } + if (message.transaction) { + userPromise.setError(Error{ResultOperationNotSupported, + "transactions are not implemented yet in the scalable-topics client"}); + return userFuture; + } + + // Register in the in-flight set (for flush) BEFORE dispatching: a warm send can complete + // synchronously, which would otherwise run the removal listener before insertion. + std::uint64_t inFlightId = 0; + { + std::lock_guard lock(mutex_); + inFlightId = nextInFlightId_++; + inFlight_.insert_or_assign(inFlightId, userFuture); + } + std::weak_ptr weak = weak_from_this(); + userFuture.addListener([weak, inFlightId](const Expected&) { + if (auto self = weak.lock()) { + std::lock_guard lock(self->mutex_); + self->inFlight_.erase(inFlightId); + } + }); + + auto layout = snapshotLayout(); + Expected route = + message.key.has_value() ? router_.route(*message.key, *layout) : router_.routeRoundRobin(*layout); + if (!route) { + // Route failure (no active segment yet / uncovered hash) is terminal, not retried. + userPromise.setError(route.error()); + return userFuture; + } + dispatchSend(std::move(message), *route, /*attempt*/ 0, userPromise); + return userFuture; +} + +void StProducerImpl::dispatchSend(OutgoingMessage message, std::uint64_t segmentId, int attempt, + const detail::Promise& userPromise) { + auto self = shared_from_this(); + getOrCreateSegmentProducerAsync(segmentId).addListener( + [self, message = std::move(message), segmentId, attempt, + userPromise](const Expected& created) mutable { + if (!created) { + self->handleSegmentFailure(created.error(), std::move(message), segmentId, attempt, + userPromise); + return; + } + pulsar::Producer producer = *created; + pulsar::Message classicMessage = self->buildClassicMessage(message); + producer.sendAsync( + classicMessage, [self, message = std::move(message), segmentId, attempt, userPromise]( + pulsar::Result result, const pulsar::MessageId& v4Id) mutable { + if (result == pulsar::ResultOk) { + userPromise.setValue( + MessageIdFactory::create(v4Id, static_cast(segmentId))); + } else { + self->handleSegmentFailure(Error{result, ""}, std::move(message), segmentId, attempt, + userPromise); + } + }); + }); +} + +void StProducerImpl::handleSegmentFailure(Error error, OutgoingMessage message, std::uint64_t segmentId, + int attempt, const detail::Promise& userPromise) { + if (!isSegmentGoneError(error.result, error.message) || attempt >= kSendRetryMaxAttempts || + closed_.load()) { + userPromise.setError(std::move(error)); + return; + } + // The segment was sealed/terminated under us. Drop its cached producer so the retry recreates it + // against the layout the DAG watch is about to deliver, then re-route after a short backoff. + { + std::lock_guard lock(mutex_); + segmentProducers_.erase(segmentId); + } + LOG_INFO("[" << topic_ << "] segment " << segmentId << " is gone; retrying send, attempt " + << (attempt + 1) << " of " << kSendRetryMaxAttempts); + auto timer = classic_->getIOExecutorProvider()->get()->createDeadlineTimer(); + const std::int64_t delayMs = std::min(100 * (attempt + 1), kSendRetryMaxBackoffMs); + timer->expires_from_now(std::chrono::milliseconds(delayMs)); + auto self = shared_from_this(); + timer->async_wait([self, message = std::move(message), attempt, userPromise, + timer](const ASIO_ERROR& ec) mutable { + if (ec || self->closed_.load()) { + userPromise.setError(Error{ResultAlreadyClosed, "producer closed during send retry"}); + return; + } + auto layout = self->snapshotLayout(); + Expected route = message.key.has_value() ? self->router_.route(*message.key, *layout) + : self->router_.routeRoundRobin(*layout); + if (!route) { + userPromise.setError(route.error()); + return; + } + self->dispatchSend(std::move(message), *route, attempt + 1, userPromise); + }); +} + +pulsar::Message StProducerImpl::buildClassicMessage(const OutgoingMessage& message) const { + pulsar::MessageBuilder builder; + // The classic client copies the content synchronously, so publishing the borrowed view here + // still honors the zero-copy (BytesView) lifetime contract. + if (message.usesView) { + builder.setContent(message.payloadView.data(), message.payloadView.size()); + } else { + builder.setContent(message.payload.data(), message.payload.size()); + } + if (message.key) builder.setPartitionKey(*message.key); + for (const auto& [key, value] : message.properties) builder.setProperty(key, value); + if (message.eventTime) { + builder.setEventTimestamp(static_cast( + std::chrono::duration_cast(message.eventTime->time_since_epoch()) + .count())); + } + if (message.sequenceId) builder.setSequenceId(*message.sequenceId); + if (message.deliverAt) { + builder.setDeliverAt(static_cast( + std::chrono::duration_cast(message.deliverAt->time_since_epoch()) + .count())); + } + if (!message.replicationClusters.empty()) builder.setReplicationClusters(message.replicationClusters); + return builder.build(); +} + +std::optional StProducerImpl::lastSequenceId() const { + std::int64_t max = config_.initialSequenceId.value_or(-1); + std::lock_guard lock(mutex_); + for (const auto& [segmentId, future] : segmentProducers_) { + if (future.isReady()) { + auto result = future.get(); + if (result) { + max = std::max(max, result->getLastSequenceId()); + } + } + } + return max < 0 ? std::nullopt : std::optional(max); +} + +Future StProducerImpl::flushAsync() { + std::vector> pending; + std::vector> producers; + { + std::lock_guard lock(mutex_); + pending.reserve(inFlight_.size()); + for (const auto& [id, future] : inFlight_) pending.push_back(future); + producers.reserve(segmentProducers_.size()); + for (const auto& [id, future] : segmentProducers_) producers.push_back(future); + } + + detail::Promise promise; + // +1 base count so an empty snapshot completes immediately. + auto remaining = std::make_shared>(static_cast(pending.size()) + 1); + auto reportedError = std::make_shared>(false); + auto finishOne = [promise, remaining]() { + if (remaining->fetch_sub(1) == 1) promise.setSuccess(); // no-op if already errored + }; + + // Nudge the ready classic producers so any batched sends complete; the actual completion is + // tracked by the in-flight user futures below. + for (auto& producerFuture : producers) { + if (producerFuture.isReady()) { + auto result = producerFuture.get(); + if (result) { + pulsar::Producer producer = *result; + producer.flushAsync([](pulsar::Result) {}); + } + } + } + for (auto& future : pending) { + future.addListener([promise, reportedError, finishOne](const Expected& result) { + if (!result && !reportedError->exchange(true)) promise.setError(result.error()); + finishOne(); + }); + } + finishOne(); + return promise.getFuture(); +} + +Future StProducerImpl::closeAsync() { + if (closed_.exchange(true)) { + detail::Promise promise; + promise.setSuccess(); // idempotent: already closed + return promise.getFuture(); + } + if (dagWatch_) dagWatch_->close(); + + std::vector> producers; + { + std::lock_guard lock(mutex_); + producers.reserve(segmentProducers_.size()); + for (auto& [id, future] : segmentProducers_) producers.push_back(future); + segmentProducers_.clear(); + } + + detail::Promise promise; + auto remaining = std::make_shared>(static_cast(producers.size()) + 1); + auto finishOne = [promise, remaining]() { + if (remaining->fetch_sub(1) == 1) promise.setSuccess(); + }; + for (auto& producerFuture : producers) { + producerFuture.addListener([finishOne](const Expected& result) { + if (result) { + pulsar::Producer producer = *result; + producer.closeAsync( + [finishOne](pulsar::Result) { finishOne(); }); // swallow per-producer errors + } else { + finishOne(); // creation itself failed — nothing to close + } + }); + } + finishOne(); + return promise.getFuture(); +} + +std::shared_ptr StProducerImpl::snapshotLayout() const { + std::lock_guard lock(mutex_); + return currentLayout_; +} + +bool StProducerImpl::isSegmentGoneError(pulsar::Result result, const std::string& message) { + if (result == pulsar::ResultTopicTerminated || result == pulsar::ResultAlreadyClosed) return true; + // Some paths surface a sealed segment as a generic error carrying the terminated text. + return message.find("TopicTerminated") != std::string::npos || + message.find("terminated") != std::string::npos; +} + +bool StProducerImpl::requiresExclusiveAttach(ProducerAccessMode mode) { + return mode != ProducerAccessMode::Shared; +} + +pulsar::ProducerConfiguration::ProducerAccessMode StProducerImpl::toClassicAccessMode( + ProducerAccessMode mode) { + // The st and classic enums do NOT share numeric values — map by name. + switch (mode) { + case ProducerAccessMode::Shared: + return pulsar::ProducerConfiguration::Shared; + case ProducerAccessMode::Exclusive: + return pulsar::ProducerConfiguration::Exclusive; + case ProducerAccessMode::ExclusiveWithFencing: + return pulsar::ProducerConfiguration::ExclusiveWithFencing; + case ProducerAccessMode::WaitForExclusive: + return pulsar::ProducerConfiguration::WaitForExclusive; + } + return pulsar::ProducerConfiguration::Shared; +} + +} // namespace pulsar::st diff --git a/lib/st/StProducerImpl.h b/lib/st/StProducerImpl.h new file mode 100644 index 00000000..24a2db02 --- /dev/null +++ b/lib/st/StProducerImpl.h @@ -0,0 +1,134 @@ +/** + * 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 "ProducerImplBase.h" +#include "SegmentLayout.h" +#include "lib/ClientImpl.h" + +namespace pulsar::st { + +/** + * The scalable-topics producer: a faithful port of the Java v5 ScalableTopicProducer. + * + * A scalable topic's segments are ordinary persistent topics, so this owns ONE + * DagWatchSession per topic and, per active segment, a classic pulsar::Producer + * pinned to the DAG-provided owner broker (created lazily on first send, or eagerly + * for exclusive access modes). Each publish is routed by key to a segment through + * the SegmentRouter, the classic producer publishes it, and the returned classic + * MessageId is wrapped with the segment id into a pulsar::st::MessageId. When a + * segment is sealed/terminated under the producer, the send retries with backoff and + * re-routes onto the new layout the DAG watch delivers. + * + * Unlike Java, no per-segment dispatch chain is needed: the C++ Future fires its + * listeners in registration (FIFO) order, so multiple sends waiting on one segment's + * creation future dispatch in call order for free. + */ +class StProducerImpl final : public ProducerImplBase, public std::enable_shared_from_this { + public: + StProducerImpl(pulsar::ClientImplPtr classic, ProducerConfig config); + + /** + * Start the DAG watch and become ready. The returned future completes once the + * initial layout arrives (and, for exclusive access modes, once every active + * segment's producer is attached), or fails if the lookup/attach fails. + */ + Future start(); + + // ProducerImplBase + Future sendAsync(OutgoingMessage message) override; + std::string_view topic() const override { return topic_; } + std::string_view producerName() const override { return producerName_; } + std::optional lastSequenceId() const override; + Future flushAsync() override; + Future closeAsync() override; + + private: + friend struct StProducerTestAccess; // broker-free access to the pure config/routing helpers + + // Number of send attempts once a segment is gone (seal/terminate), and the cap on + // the per-attempt backoff — matches the Java v5 producer. + static constexpr int kSendRetryMaxAttempts = 10; + static constexpr std::int64_t kSendRetryMaxBackoffMs = 500; + + // Build a FRESH classic producer configuration for one segment. Never copy-then-mutate + // a shared base: pulsar::ProducerConfiguration's copy constructor shares its impl, so a + // per-segment copy would clobber every sibling. + pulsar::ProducerConfiguration buildSegmentConfiguration(const Segment& segment) const; + + Future getOrCreateSegmentProducerAsync(std::uint64_t segmentId); + + // start()'s future handler: only surfaces a start-time failure. The success path (apply + // the initial layout and complete startPromise_) is driven by the layout listener, which + // the DagWatchSession invokes right after completing start()'s future. + void onStartResult(const Expected& result); + // The layout-change listener: fires for every accepted layout (the first has an empty + // oldLayout). Swaps the current layout, retires producers for departed segments, and on + // the first call completes startPromise_ (eagerly attaching all segments for exclusive + // modes). + void onLayoutChange(const SegmentLayout& newLayout, const SegmentLayout& oldLayout); + void completeStart(); + + void dispatchSend(OutgoingMessage message, std::uint64_t segmentId, int attempt, + const detail::Promise& userPromise); + // On a send failure: retry after backoff on a fresh layout if the segment is gone and the + // attempt budget remains, otherwise fail the user's future. + void handleSegmentFailure(Error error, OutgoingMessage message, std::uint64_t segmentId, int attempt, + const detail::Promise& userPromise); + pulsar::Message buildClassicMessage(const OutgoingMessage& message) const; + + std::shared_ptr snapshotLayout() const; + static bool isSegmentGoneError(pulsar::Result result, const std::string& message); + static bool requiresExclusiveAttach(ProducerAccessMode mode); + static pulsar::ProducerConfiguration::ProducerAccessMode toClassicAccessMode(ProducerAccessMode mode); + + pulsar::ClientImplPtr classic_; + const ProducerConfig config_; + const std::string topic_; + const std::string producerName_; + DagWatchSessionPtr dagWatch_; + SegmentRouter router_; + detail::Promise startPromise_; + std::atomic closed_{false}; + + mutable std::mutex mutex_; + bool sawFirstLayout_ = false; // guarded by mutex_ + std::shared_ptr currentLayout_; // guarded by mutex_ (never null) + std::unordered_map> segmentProducers_; // guarded by mutex_ + std::unordered_map> inFlight_; // guarded by mutex_ (for flush) + std::uint64_t nextInFlightId_ = 0; // guarded by mutex_ +}; + +using StProducerImplPtr = std::shared_ptr; + +} // namespace pulsar::st diff --git a/tests/st/FutureTest.cc b/tests/st/FutureTest.cc index 316eaf1b..47e660de 100644 --- a/tests/st/FutureTest.cc +++ b/tests/st/FutureTest.cc @@ -27,6 +27,7 @@ #include #include #include +#include using namespace pulsar::st; using pulsar::st::detail::Promise; @@ -82,6 +83,33 @@ TEST(FutureTest, testListenerAfterCompletionRunsSynchronously) { ASSERT_EQ(seen, 9); } +// The scalable-topics producer relies on listeners firing in registration order to +// preserve send ordering to a segment without a per-segment dispatch chain (which +// Java needs only because JDK CompletableFuture gives no fire-order guarantee). Lock +// that guarantee in here. +TEST(FutureTest, testListenersFireInRegistrationOrder) { + Promise promise; + Future future = promise.getFuture(); + std::vector order; + for (int i = 0; i < 5; i++) { + future.addListener([&order, i](const Expected&) { order.push_back(i); }); + } + promise.setValue(1); + ASSERT_EQ(order, (std::vector{0, 1, 2, 3, 4})); +} + +TEST(FutureTest, testListenerAddedAfterCompletionRunsAfterEarlierOnes) { + Promise promise; + Future future = promise.getFuture(); + std::vector order; + future.addListener([&order](const Expected&) { order.push_back(0); }); + promise.setValue(1); // fires listener 0 + // A listener registered after completion runs synchronously, and after the ones + // that were already registered — so a late same-segment send never jumps the queue. + future.addListener([&order](const Expected&) { order.push_back(1); }); + ASSERT_EQ(order, (std::vector{0, 1})); +} + TEST(FutureTest, testCompleteIsFirstWriterWins) { Promise promise; ASSERT_TRUE(promise.setValue(1)); diff --git a/tests/st/StClientTest.cc b/tests/st/StClientTest.cc index 839b7ee1..7c67a917 100644 --- a/tests/st/StClientTest.cc +++ b/tests/st/StClientTest.cc @@ -73,17 +73,6 @@ TEST(StClientTest, testBuildWithPoliciesWithoutBroker) { ASSERT_TRUE(client->close()); } -TEST(StClientTest, testProducerCreationReportsNotSupportedYet) { - auto client = PulsarClient::builder().serviceUrl("pulsar://localhost:6650").build(); - ASSERT_TRUE(client); - - auto producer = client->newProducer().topic("st-topic").create(); - ASSERT_FALSE(producer); - ASSERT_EQ(producer.error().result, ResultOperationNotSupported); - - ASSERT_TRUE(client->close()); -} - TEST(StClientTest, testTransactionReportsNotSupportedYet) { auto client = PulsarClient::builder().serviceUrl("pulsar://localhost:6650").build(); ASSERT_TRUE(client); diff --git a/tests/st/StMessageIdTest.cc b/tests/st/StMessageIdTest.cc new file mode 100644 index 00000000..cba876d1 --- /dev/null +++ b/tests/st/StMessageIdTest.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. + */ +#include +#include + +#include +#include +#include +#include + +#include "lib/st/MessageIdImpl.h" + +using namespace pulsar::st; + +namespace { + +MessageId makeId(std::int64_t segmentId, std::int64_t ledgerId, std::int64_t entryId, + std::int32_t batchIndex = -1, std::int32_t partition = -1) { + return MessageIdFactory::create(pulsar::MessageId(partition, ledgerId, entryId, batchIndex), segmentId); +} + +} // namespace + +TEST(StMessageIdTest, testSentinels) { + ASSERT_TRUE(static_cast(MessageId::earliest())); + ASSERT_TRUE(static_cast(MessageId::latest())); + ASSERT_TRUE(MessageId::earliest() == MessageId::earliest()); + ASSERT_FALSE(MessageId::earliest() == MessageId::latest()); + ASSERT_TRUE(MessageId::earliest() < MessageId::latest()); +} + +TEST(StMessageIdTest, testOrderingWithinAndAcrossSegments) { + MessageId a = makeId(1, 10, 1); + MessageId b = makeId(1, 10, 2); + MessageId c = makeId(2, 5, 0); + + // Same segment: classic id order decides. + ASSERT_TRUE(a < b); + ASSERT_TRUE(b > a); + // Different segments: segment id decides, regardless of ledger/entry. + ASSERT_TRUE(b < c); + ASSERT_TRUE(a <= a); + ASSERT_TRUE(a == makeId(1, 10, 1)); + ASSERT_FALSE(a == b); + ASSERT_FALSE(a == c); +} + +TEST(StMessageIdTest, testEmptyIdComparisons) { + MessageId empty1; + MessageId empty2; + MessageId real = makeId(0, 1, 1); + ASSERT_TRUE(empty1 == empty2); + ASSERT_FALSE(empty1 == real); + ASSERT_TRUE(empty1 < real); // empty sorts first +} + +TEST(StMessageIdTest, testHashSupportsUnorderedContainers) { + std::unordered_set ids; + ids.insert(makeId(1, 10, 1)); + ids.insert(makeId(1, 10, 2)); + ids.insert(makeId(2, 10, 1)); + ids.insert(makeId(1, 10, 1)); // duplicate + ASSERT_EQ(ids.size(), 3u); + ASSERT_EQ(std::hash()(makeId(1, 10, 1)), std::hash()(makeId(1, 10, 1))); +} + +TEST(StMessageIdTest, testStreamOutput) { + std::ostringstream out; + out << makeId(7, 10, 3); + ASSERT_NE(out.str().find("segment=7"), std::string::npos); + + std::ostringstream emptyOut; + emptyOut << MessageId(); + ASSERT_EQ(emptyOut.str(), "{empty}"); +} + +TEST(StMessageIdTest, testByteArrayRoundtripProducerPath) { + MessageId original = makeId(42, 1234, 5678, 3, 5); + std::vector bytes = original.toByteArray(); + ASSERT_FALSE(bytes.empty()); + + MessageId restored = MessageId::fromByteArray(std::span(bytes)); + ASSERT_TRUE(static_cast(restored)); + ASSERT_TRUE(restored == original); + + const auto& impl = MessageIdFactory::impl(restored); + ASSERT_EQ(impl->segmentId, 42); + ASSERT_EQ(impl->v4MessageId.ledgerId(), 1234); + ASSERT_EQ(impl->v4MessageId.entryId(), 5678); + ASSERT_EQ(impl->v4MessageId.batchIndex(), 3); + ASSERT_EQ(impl->v4MessageId.partition(), 5); + ASSERT_TRUE(impl->positionVector.empty()); + ASSERT_FALSE(impl->parentTopic.has_value()); + ASSERT_FALSE(impl->multiTopicVector.has_value()); +} + +TEST(StMessageIdTest, testByteArrayRoundtripFullSections) { + auto impl = std::make_shared(); + impl->v4MessageId = pulsar::MessageId(-1, 10, 20, -1); + impl->segmentId = 3; + impl->positionVector.emplace(1, pulsar::MessageId(-1, 100, 1, -1)); + impl->positionVector.emplace(2, pulsar::MessageId(-1, 200, 2, -1)); + impl->parentTopic = "topic://public/default/orders"; + std::map> multi; + multi["topic://public/default/other"].emplace(9, pulsar::MessageId(-1, 900, 9, -1)); + impl->multiTopicVector = multi; + + MessageId original = MessageIdFactory::create(impl); + auto bytes = original.toByteArray(); + MessageId restored = MessageId::fromByteArray(std::span(bytes)); + ASSERT_TRUE(static_cast(restored)); + ASSERT_TRUE(restored == original); + + const auto& r = MessageIdFactory::impl(restored); + ASSERT_EQ(r->positionVector.size(), 2u); + ASSERT_EQ(r->positionVector.at(1).ledgerId(), 100); + ASSERT_EQ(r->positionVector.at(2).ledgerId(), 200); + ASSERT_TRUE(r->parentTopic.has_value()); + ASSERT_EQ(*r->parentTopic, "topic://public/default/orders"); + ASSERT_TRUE(r->multiTopicVector.has_value()); + ASSERT_EQ(r->multiTopicVector->size(), 1u); + ASSERT_EQ(r->multiTopicVector->at("topic://public/default/other").at(9).ledgerId(), 900); +} + +TEST(StMessageIdTest, testMalformedBytesYieldEmptyId) { + ASSERT_FALSE(static_cast(MessageId::fromByteArray({}))); + + std::vector tooShort(6, std::byte{1}); + ASSERT_FALSE(static_cast(MessageId::fromByteArray(std::span(tooShort)))); + + std::vector garbage(64, std::byte{0xFF}); + ASSERT_FALSE(static_cast(MessageId::fromByteArray(std::span(garbage)))); + + // A truncated valid id must not parse. + auto bytes = makeId(1, 2, 3).toByteArray(); + bytes.resize(bytes.size() / 2); + ASSERT_FALSE(static_cast(MessageId::fromByteArray(std::span(bytes)))); +} + +TEST(StMessageIdTest, testEmptyIdSerializesToNothing) { ASSERT_TRUE(MessageId().toByteArray().empty()); } diff --git a/tests/st/StProducerTest.cc b/tests/st/StProducerTest.cc new file mode 100644 index 00000000..9af869f5 --- /dev/null +++ b/tests/st/StProducerTest.cc @@ -0,0 +1,118 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include +#include +#include + +#include +#include + +#include "lib/st/MessageIdImpl.h" +#include "lib/st/StProducerImpl.h" + +namespace pulsar::st { + +// Test-only access to StProducerImpl's private, broker-independent helpers. These read +// only config_ and the segment argument (never the classic client), so they can run with +// a null client and no connection. +struct StProducerTestAccess { + static pulsar::ProducerConfiguration segmentConfig(StProducerImpl& producer, const Segment& segment) { + return producer.buildSegmentConfiguration(segment); + } + static pulsar::ProducerConfiguration::ProducerAccessMode toClassic(ProducerAccessMode mode) { + return StProducerImpl::toClassicAccessMode(mode); + } + static bool segmentGone(pulsar::Result result, const std::string& message) { + return StProducerImpl::isSegmentGoneError(result, message); + } +}; + +namespace { + +ProducerConfig baseConfig() { + ProducerConfig config; + config.topic = "topic://public/default/orders"; + config.producerName = "orders-producer"; + return config; +} + +Segment normalSegment(std::uint64_t id) { + Segment segment; + segment.segmentId = id; + segment.segmentTopicName = "segment://public/default/orders/0000-ffff-" + std::to_string(id); + return segment; +} + +Segment legacySegment(std::uint64_t id) { + Segment segment = normalSegment(id); + segment.legacyTopicName = "persistent://public/default/orders-partition-" + std::to_string(id); + return segment; +} + +// Guards the ProducerConfiguration shallow-copy defect: pulsar::ProducerConfiguration's copy +// constructor shares its impl, so building a second segment's config by copy-then-mutate would +// clobber the first. Each segment must get an independent, freshly-built config. +TEST(StProducerTest, testPerSegmentConfigsAreIsolated) { + StProducerImpl producer(nullptr, baseConfig()); + + auto confA = StProducerTestAccess::segmentConfig(producer, normalSegment(3)); + auto confB = StProducerTestAccess::segmentConfig(producer, legacySegment(7)); + + // If the configs shared state, building B would rename A too. + EXPECT_EQ(confA.getProducerName(), "orders-producer-seg-3"); + EXPECT_EQ(confB.getProducerName(), "orders-producer-seg-7"); + + // The legacy managed-metadata marker belongs to the legacy segment only. + EXPECT_EQ(confB.getProperties().count("__pulsar.v5.managed"), 1u); + EXPECT_EQ(confA.getProperties().count("__pulsar.v5.managed"), 0u); +} + +// The st and classic ProducerAccessMode enums do NOT share numeric values (st orders +// ExclusiveWithFencing before WaitForExclusive; classic is the reverse), so a cast would be +// wrong — the mapping must go by name. +TEST(StProducerTest, testAccessModeMappedByName) { + using Classic = pulsar::ProducerConfiguration; + EXPECT_EQ(StProducerTestAccess::toClassic(ProducerAccessMode::Shared), Classic::Shared); + EXPECT_EQ(StProducerTestAccess::toClassic(ProducerAccessMode::Exclusive), Classic::Exclusive); + EXPECT_EQ(StProducerTestAccess::toClassic(ProducerAccessMode::ExclusiveWithFencing), + Classic::ExclusiveWithFencing); + EXPECT_EQ(StProducerTestAccess::toClassic(ProducerAccessMode::WaitForExclusive), + Classic::WaitForExclusive); +} + +TEST(StProducerTest, testSegmentGoneClassification) { + EXPECT_TRUE(StProducerTestAccess::segmentGone(ResultTopicTerminated, "")); + EXPECT_TRUE(StProducerTestAccess::segmentGone(ResultAlreadyClosed, "")); + // A not-ready layout is a transient routing failure, NOT a gone segment — treating it as + // gone would spin the retry loop instead of failing terminally. + EXPECT_FALSE(StProducerTestAccess::segmentGone(ResultServiceUnitNotReady, "")); + EXPECT_FALSE(StProducerTestAccess::segmentGone(ResultConnectError, "")); +} + +TEST(StProducerTest, testMintedMessageIdCarriesSegmentId) { + MessageId id = MessageIdFactory::create(pulsar::MessageId::earliest(), 42); + ASSERT_TRUE(static_cast(id)); + const auto& impl = MessageIdFactory::impl(id); + ASSERT_TRUE(impl); + EXPECT_EQ(impl->segmentId, 42); + EXPECT_NE(impl->segmentId, MessageIdImpl::kNoSegment); +} + +} // namespace +} // namespace pulsar::st diff --git a/tests/st/StTopicNameTest.cc b/tests/st/StTopicNameTest.cc new file mode 100644 index 00000000..1e3f4a5b --- /dev/null +++ b/tests/st/StTopicNameTest.cc @@ -0,0 +1,90 @@ +/** + * 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 "lib/TopicName.h" + +// The segment:// domain — the per-segment backing topics of a scalable topic — +// must parse in the classic TopicName (the per-segment producers attach to +// them), while every classic user-facing entry point rejects it: only the +// pulsar::st client may touch segment topics. + +using namespace pulsar; + +static const std::string kSegmentTopic = "segment://public/default/orders/0000-7fff-1"; + +TEST(StTopicNameTest, testSegmentTopicParses) { + auto topicName = TopicName::get(kSegmentTopic); + ASSERT_TRUE(topicName); + ASSERT_EQ(topicName->getDomain(), "segment"); + ASSERT_EQ(topicName->getProperty(), "public"); + ASSERT_EQ(topicName->getNamespacePortion(), "default"); + // The local name keeps the '/' shape. + ASSERT_EQ(topicName->getLocalName(), "orders/0000-7fff-1"); + ASSERT_TRUE(topicName->isV2Topic()); + ASSERT_EQ(topicName->toString(), kSegmentTopic); + // A segment topic is the persistent backing topic of one segment. + ASSERT_TRUE(topicName->isPersistent()); + ASSERT_TRUE(topicName->isSegment()); + // Never partitioned. + ASSERT_EQ(topicName->getPartitionIndex(), -1); +} + +TEST(StTopicNameTest, testSegmentTopicRequiresDescriptor) { + // Missing the '' path component. + ASSERT_FALSE(TopicName::get("segment://public/default/orders")); + ASSERT_FALSE(TopicName::get("segment://public/default")); +} + +TEST(StTopicNameTest, testClassicTopicsUnaffected) { + auto v2 = TopicName::get("persistent://public/default/orders"); + ASSERT_TRUE(v2); + ASSERT_TRUE(v2->isPersistent()); + ASSERT_FALSE(v2->isSegment()); + + // A 5-component persistent name still parses as the legacy V1 format. + auto v1 = TopicName::get("persistent://tenant/cluster/ns/orders"); + ASSERT_TRUE(v1); + ASSERT_FALSE(v1->isV2Topic()); + ASSERT_EQ(v1->getCluster(), "cluster"); + ASSERT_FALSE(v1->isSegment()); + + // topic:// (the scalable-topic identity scheme) stays out of the classic + // TopicName: the pulsar::st client resolves it before reaching here. + ASSERT_FALSE(TopicName::get("topic://public/default/orders")); +} + +// The classic user-facing entry points must reject segment topics before any +// network activity, so these run without a broker. + +TEST(StTopicNameTest, testClassicApiRejectsSegmentTopics) { + Client client("pulsar://localhost:6650"); + + Producer producer; + ASSERT_EQ(client.createProducer(kSegmentTopic, producer), ResultInvalidTopicName); + + Consumer consumer; + ASSERT_EQ(client.subscribe(kSegmentTopic, "sub", consumer), ResultInvalidTopicName); + + Reader reader; + ASSERT_EQ(client.createReader(kSegmentTopic, MessageId::earliest(), {}, reader), ResultInvalidTopicName); + + client.close(); +}