[Issue 11496][C++] Allow partitioned producers to start lazily

pulsar-client-cpp/include/pulsar/ProducerConfiguration.h

@@ -262,6 +262,27 @@ class PULSAR_PUBLIC ProducerConfiguration {
      */
     HashingScheme getHashingScheme() const;
 
+    /**
+     * This config affects producers of partitioned topics only. It controls whether
+     * producers register and connect immediately to the owner broker of each partition
+     * or start lazily on demand. The internal producer of one partition is always
+     * started eagerly, chosen by the routing policy, but the internal producers of
+     * any additional partitions are started on demand, upon receiving their first
+     * message.
+     * Using this mode can reduce the strain on brokers for topics with large numbers of
+     * partitions and when the SinglePartition routing policy is used without keyed messages.
+     * Because producer connection can be on demand, this can produce extra send latency
+     * for the first messages of a given partition.
+     * @param true/false as to whether to start partition producers lazily
+     * @return
+     */
+    ProducerConfiguration& setLazyStartPartitionedProducers(bool);
+
+    /**
+     * The getter associated with setLazyStartPartitionedProducers()
+     */
+    bool getLazyStartPartitionedProducers() const;
+
     /**
      * The setter associated with getBlockIfQueueFull()
      */

pulsar-client-cpp/include/pulsar/c/producer_configuration.h

@@ -144,6 +144,12 @@ PULSAR_PUBLIC void pulsar_producer_configuration_set_hashing_scheme(pulsar_produ
 PULSAR_PUBLIC pulsar_hashing_scheme
 pulsar_producer_configuration_get_hashing_scheme(pulsar_producer_configuration_t *conf);
 
+PULSAR_PUBLIC void pulsar_producer_configuration_set_lazy_start_partitioned_producers(
+    pulsar_producer_configuration_t *conf, int useLazyStartPartitionedProducers);
+
+PULSAR_PUBLIC int pulsar_producer_configuration_get_lazy_start_partitioned_producers(
+    pulsar_producer_configuration_t *conf);
+
 PULSAR_PUBLIC void pulsar_producer_configuration_set_block_if_queue_full(
     pulsar_producer_configuration_t *conf, int blockIfQueueFull);
 

pulsar-client-cpp/lib/HandlerBase.cc

@@ -35,14 +35,23 @@ HandlerBase::HandlerBase(const ClientImplPtr& client, const std::string& topic,
       mutex_(),
       creationTimestamp_(TimeUtils::now()),
       operationTimeut_(seconds(client->conf().getOperationTimeoutSeconds())),
-      state_(Pending),
+      state_(NotStarted),
       backoff_(backoff),
       epoch_(0),
       timer_(executor_->createDeadlineTimer()) {}
 
 HandlerBase::~HandlerBase() { timer_->cancel(); }
 
-void HandlerBase::start() { grabCnx(); }
+void HandlerBase::start() {
+    Lock lock(mutex_);
+    // guard against concurrent state changes such as closing
+    if (state_ == NotStarted) {
+        state_ = Pending;
+        lock.unlock();
+
+        grabCnx();
+    }
+}
 
 void HandlerBase::grabCnx() {
     Lock lock(mutex_);
@@ -106,6 +115,7 @@ void HandlerBase::handleDisconnection(Result result, ClientConnectionWeakPtr con
             scheduleReconnection(handler);
             break;
 
+        case NotStarted:
         case Closing:
         case Closed:
         case Failed:

pulsar-client-cpp/lib/HandlerBase.h

@@ -97,6 +97,7 @@ class HandlerBase {
 
     enum State
     {
+        NotStarted,
         Pending,
         Ready,
         Closing,

pulsar-client-cpp/lib/PartitionedProducerImpl.cc

@@ -87,13 +87,18 @@ unsigned int PartitionedProducerImpl::getNumPartitionsWithLock() const {
     return getNumPartitions();
 }
 
-ProducerImplPtr PartitionedProducerImpl::newInternalProducer(unsigned int partition) const {
+ProducerImplPtr PartitionedProducerImpl::newInternalProducer(unsigned int partition, bool lazy) {
     using namespace std::placeholders;
     std::string topicPartitionName = topicName_->getTopicPartitionName(partition);
     auto producer = std::make_shared<ProducerImpl>(client_, topicPartitionName, conf_, partition);
-    producer->getProducerCreatedFuture().addListener(
-        std::bind(&PartitionedProducerImpl::handleSinglePartitionProducerCreated,
-                  const_cast<PartitionedProducerImpl*>(this)->shared_from_this(), _1, _2, partition));
+
+    if (lazy) {
+        createLazyPartitionProducer(partition);
+    } else {
+        producer->getProducerCreatedFuture().addListener(
+            std::bind(&PartitionedProducerImpl::handleSinglePartitionProducerCreated,
+                      const_cast<PartitionedProducerImpl*>(this)->shared_from_this(), _1, _2, partition));
+    }
 
     LOG_DEBUG("Creating Producer for single Partition - " << topicPartitionName);
     return producer;
@@ -104,12 +109,28 @@ void PartitionedProducerImpl::start() {
     // create producer per partition
     // Here we don't need `producersMutex` to protect `producers_`, because `producers_` can only be increased
     // when `state_` is Ready
-    for (unsigned int i = 0; i < getNumPartitions(); i++) {
-        producers_.push_back(newInternalProducer(i));
-    }
 
-    for (ProducerList::const_iterator prod = producers_.begin(); prod != producers_.end(); prod++) {
-        (*prod)->start();
+    if (conf_.getLazyStartPartitionedProducers()) {
+        // start one producer now, to ensure authz errors occur now
+        // if the SinglePartition router is used, then this producer will serve
+        // all non-keyed messages in the future
+        Message msg = MessageBuilder().setContent("x").build();
+        short partition = (short)(routerPolicy_->getPartition(msg, *topicMetadata_));
+
+        for (unsigned int i = 0; i < getNumPartitions(); i++) {
+            bool lazy = (short)i != partition;
+            producers_.push_back(newInternalProducer(i, lazy));
+        }
+
+        producers_[partition]->start();
+    } else {
+        for (unsigned int i = 0; i < getNumPartitions(); i++) {
+            producers_.push_back(newInternalProducer(i, false));
+        }
+
+        for (ProducerList::const_iterator prod = producers_.begin(); prod != producers_.end(); prod++) {
+            (*prod)->start();
+        }
     }
 }
 
@@ -147,8 +168,27 @@ void PartitionedProducerImpl::handleSinglePartitionProducerCreated(Result result
     }
 }
 
+void PartitionedProducerImpl::createLazyPartitionProducer(unsigned int partitionIndex) {
+    const auto numPartitions = getNumPartitions();
+    assert(numProducersCreated_ <= numPartitions);
+    assert(partitionIndex <= numPartitions);
+    numProducersCreated_++;
+    if (numProducersCreated_ == numPartitions) {
+        state_ = Ready;
+        if (partitionsUpdateTimer_) {
+            runPartitionUpdateTask();
+        }
+        partitionedProducerCreatedPromise_.setValue(shared_from_this());
+    }
+}
+
 // override
 void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callback) {
+    if (!assertState(Ready)) {
+        callback(ResultAlreadyClosed, msg.getMessageId());
+        return;
+    }
+
     // get partition for this message from router policy
     Lock producersLock(producersMutex_);
     short partition = (short)(routerPolicy_->getPartition(msg, *topicMetadata_));
@@ -161,7 +201,14 @@ void PartitionedProducerImpl::sendAsync(const Message& msg, SendCallback callbac
     }
     // find a producer for that partition, index should start from 0
     ProducerImplPtr producer = producers_[partition];
+
+    // if the producer is not started (lazy producer), then kick-off the start process
+    if (!producer->isStarted()) {
+        producer->start();
+    }
+
     producersLock.unlock();
+
     // send message on that partition
     producer->sendAsync(msg, callback);
 }
@@ -175,6 +222,11 @@ void PartitionedProducerImpl::setState(const PartitionedProducerState state) {
     lock.unlock();
 }
 
+bool PartitionedProducerImpl::assertState(const PartitionedProducerState state) {
+    Lock lock(mutex_);
+    return state_ == state;
+}
+
 const std::string& PartitionedProducerImpl::getProducerName() const {
     Lock producersLock(producersMutex_);
     return producers_[0]->getProducerName();
@@ -285,7 +337,9 @@ bool PartitionedProducerImpl::isClosed() { return state_ == Closed; }
 void PartitionedProducerImpl::triggerFlush() {
     Lock producersLock(producersMutex_);
     for (ProducerList::const_iterator prod = producers_.begin(); prod != producers_.end(); prod++) {
-        (*prod)->triggerFlush();
+        if ((*prod)->isStarted()) {
+            (*prod)->triggerFlush();
+        }
     }
 }
 
@@ -322,7 +376,11 @@ void PartitionedProducerImpl::flushAsync(FlushCallback callback) {
     };
 
     for (ProducerList::const_iterator prod = producers_.begin(); prod != producers_.end(); prod++) {
-        (*prod)->flushAsync(subFlushCallback);
+        if ((*prod)->isStarted()) {
+            (*prod)->flushAsync(subFlushCallback);
+        } else {
+            subFlushCallback(ResultOk);
+        }
     }
 }
 
@@ -355,8 +413,11 @@ void PartitionedProducerImpl::handleGetPartitions(Result result,
             topicMetadata_.reset(new TopicMetadataImpl(newNumPartitions));
 
             for (unsigned int i = currentNumPartitions; i < newNumPartitions; i++) {
-                auto producer = newInternalProducer(i);
-                producer->start();
+                auto producer = newInternalProducer(i, conf_.getLazyStartPartitionedProducers());
+
+                if (!conf_.getLazyStartPartitionedProducers()) {
+                    producer->start();
+                }
                 producers_.push_back(producer);
             }
             // `runPartitionUpdateTask()` will be called in `handleSinglePartitionProducerCreated()`
@@ -379,8 +440,8 @@ bool PartitionedProducerImpl::isConnected() const {
     Lock producersLock(producersMutex_);
     const auto producers = producers_;
     producersLock.unlock();
-    for (const auto& producer : producers_) {
-        if (!producer->isConnected()) {
+    for (const auto& producer : producers) {
+        if (producer->isStarted() && !producer->isConnected()) {
             return false;
         }
     }

pulsar-client-cpp/lib/PartitionedProducerImpl.h

@@ -67,7 +67,7 @@ class PartitionedProducerImpl : public ProducerImplBase,
     uint64_t getNumberOfConnectedProducer() override;
     void handleSinglePartitionProducerCreated(Result result, ProducerImplBaseWeakPtr producerBaseWeakPtr,
                                               const unsigned int partitionIndex);
-
+    void createLazyPartitionProducer(const unsigned int partitionIndex);
     void handleSinglePartitionProducerClose(Result result, const unsigned int partitionIndex,
                                             CloseCallback callback);
 
@@ -104,7 +104,7 @@ class PartitionedProducerImpl : public ProducerImplBase,
     unsigned int getNumPartitions() const;
     unsigned int getNumPartitionsWithLock() const;
 
-    ProducerImplPtr newInternalProducer(unsigned int partition) const;
+    ProducerImplPtr newInternalProducer(unsigned int partition, bool lazy);
 
     MessageRoutingPolicyPtr routerPolicy_;
 
@@ -129,6 +129,7 @@ class PartitionedProducerImpl : public ProducerImplBase,
     void runPartitionUpdateTask();
     void getPartitionMetadata();
     void handleGetPartitions(const Result result, const LookupDataResultPtr& partitionMetadata);
+    bool assertState(const PartitionedProducerState state);
 };
 
 }  // namespace pulsar

pulsar-client-cpp/lib/Producer.cc

@@ -114,7 +114,7 @@ void Producer::flushAsync(FlushCallback callback) {
 void Producer::producerFailMessages(Result result) {
     if (impl_) {
         ProducerImpl* producerImpl = static_cast<ProducerImpl*>(impl_.get());
-        producerImpl->failPendingMessages(result);
+        producerImpl->failPendingMessages(result, true);
     }
 }
 

pulsar-client-cpp/lib/ProducerConfiguration.cc

@@ -204,6 +204,16 @@ ProducerConfiguration& ProducerConfiguration::addEncryptionKey(std::string key)
     return *this;
 }
 
+ProducerConfiguration& ProducerConfiguration::setLazyStartPartitionedProducers(
+    bool useLazyStartPartitionedProducers) {
+    impl_->useLazyStartPartitionedProducers = useLazyStartPartitionedProducers;
+    return *this;
+}
+
+bool ProducerConfiguration::getLazyStartPartitionedProducers() const {
+    return impl_->useLazyStartPartitionedProducers;
+}
+
 ProducerConfiguration& ProducerConfiguration::setSchema(const SchemaInfo& schemaInfo) {
     impl_->schemaInfo = schemaInfo;
     return *this;

pulsar-client-cpp/lib/ProducerConfigurationImpl.h

@@ -37,6 +37,7 @@ struct ProducerConfigurationImpl {
     ProducerConfiguration::PartitionsRoutingMode routingMode{ProducerConfiguration::UseSinglePartition};
     MessageRoutingPolicyPtr messageRouter;
     ProducerConfiguration::HashingScheme hashingScheme{ProducerConfiguration::BoostHash};
+    bool useLazyStartPartitionedProducers{false};
     bool blockIfQueueFull{false};
     bool batchingEnabled{true};
     unsigned int batchingMaxMessages{1000};