apache · merlimat · Jul 2, 2026 · Jul 2, 2026 · Jul 7, 2026 · Jul 7, 2026
diff --git 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<ProducerImplBase>;
 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
@@ -94,6 +94,15 @@ std::string generateRandomName() {
 
 typedef std::vector<std::string> 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<std::string>& 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<std::string>& 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<std::string>& assignedBrokerUrl) {
     if (!error.result) {
         ProducerImplBasePtr producer;
 
@@ -258,7 +288,10 @@ void ClientImpl::handleCreateProducer(const Error& error, const LookupDataResult
                 producer = std::make_shared<PartitionedProducerImpl>(
                     shared_from_this(), topicName, partitionMetadata->getPartitions(), conf, interceptors);
             } else {
-                producer = std::make_shared<ProducerImpl>(shared_from_this(), *topicName, conf, interceptors);
+                producer =
+                    std::make_shared<ProducerImpl>(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<TableViewImpl>(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
@@ -95,6 +95,14 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
     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<std::string>& 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<ClientImpl> {
     friend class PulsarFriend;
 
    private:
+    void createProducerAsyncImpl(const std::string& topic, const ProducerConfiguration& conf,
+                                 CreateProducerV2Callback callback, bool autoDownloadSchema,
+                                 bool allowSegmentTopic,
+                                 const std::optional<std::string>& assignedBrokerUrl = std::nullopt);
+
     void handleCreateProducer(const Error& error, const LookupDataResultPtr& partitionMetadata,
                               const TopicNamePtr& topicName, const ProducerConfiguration& conf,
-                              CreateProducerV2Callback callback);
+                              CreateProducerV2Callback callback,
+                              const std::optional<std::string>& 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
@@ -54,11 +54,13 @@ HandlerBase::~HandlerBase() {
     cancelTimer(*creationTimer_);
 }
 
-void HandlerBase::start() {
+void HandlerBase::start() { start(std::nullopt); }
+
+void HandlerBase::start(const optional<std::string>& 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
@@ -51,6 +51,13 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
 
     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<std::string>& assignedBrokerUrl);
+
     ClientConnectionWeakPtr getCnx() const;
     void setCnx(const ClientConnectionPtr& cnx);
     void resetCnx() { setCnx(nullptr); }

diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
@@ -50,14 +50,16 @@ 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<std::string>& assignedBrokerUrl)
     : HandlerBase(client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition),
                   Backoff(milliseconds(client->getClientConfig().getInitialBackoffIntervalMs()),
                           milliseconds(client->getClientConfig().getMaxBackoffIntervalMs()),
                           milliseconds(std::max(100, conf.getSendTimeout() - 100)))),
       conf_(conf),
       semaphore_(),
       partition_(partition),
+      assignedBrokerUrl_(assignedBrokerUrl),
       producerName_(conf_.getProducerName()),
       userProvidedProducerName_(false),
       producerStr_("[" + topic() + ", " + producerName_ + "] "),
@@ -1007,7 +1009,7 @@ void ProducerImpl::disconnectProducer(const optional<std::string>& 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
@@ -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<std::string>& assignedBrokerUrl = std::nullopt);
     ~ProducerImpl();
 
     // overrided methods from ProducerImplBase
@@ -177,6 +178,9 @@ class ProducerImpl : public HandlerBase, public ProducerImplBase {
     std::list<std::unique_ptr<OpSendMsg>> 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<std::string> assignedBrokerUrl_;
     std::string producerName_;
     bool userProvidedProducerName_;
     std::string producerStr_;

diff --git 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<std::mutex> 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://<tenant>/<ns>/<parent-topic>/<descriptor>: 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://<tenant>/<namespace>/<topic>/<descriptor> - "
+                << 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
@@ -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://<tenant>/<ns>/<parent-topic>/<hashStart>-<hashEnd>-<segmentId>.
+    // 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<TopicName> get(const std::string& topicName);