[C++] PIP 37: Support large message size

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

@@ -437,6 +437,52 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     int getPriorityLevel() const;
 
+    /**
+     * Consumer buffers chunk messages into memory until it receives all the chunks of the original message.
+     * While consuming chunk-messages, chunks from same message might not be contiguous in the stream and they
+     * might be mixed with other messages' chunks. so, consumer has to maintain multiple buffers to manage
+     * chunks coming from different messages. This mainly happens when multiple publishers are publishing
+     * messages on the topic concurrently or publisher failed to publish all chunks of the messages.
+     *
+     * eg: M1-C1, M2-C1, M1-C2, M2-C2
+     * Here, Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 belong to M2 message.
+     *
+     * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it
+     * can be guarded by providing this maxPendingChunkedMessage threshold. Once, consumer reaches this
+     * threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver
+     * later by marking it unacked. See setAutoOldestChunkedMessageOnQueueFull.
+     *
+     * If it's zero, the pending chunked messages will not be limited.
+     *
+     * Default: 100
+     *
+     * @param maxPendingChunkedMessage the number of max pending chunked messages
+     */
+    ConsumerConfiguration& setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage);
+
+    /**
+     * The associated getter of setMaxPendingChunkedMessage
+     */
+    size_t getMaxPendingChunkedMessage() const;
+
+    /**
+     * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it
+     * can be guarded by providing the maxPendingChunkedMessage threshold. See setMaxPendingChunkedMessage.
+     * Once, consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acking
+     * if autoAckOldestChunkedMessageOnQueueFull is true else it marks them for redelivery.
+     *
+     * Default: false
+     *
+     * @param autoAckOldestChunkedMessageOnQueueFull whether to ack the discarded chunked message
+     */
+    ConsumerConfiguration& setAutoOldestChunkedMessageOnQueueFull(
+        bool autoAckOldestChunkedMessageOnQueueFull);
+
+    /**
+     * The associated getter of setAutoOldestChunkedMessageOnQueueFull
+     */
+    bool isAutoOldestChunkedMessageOnQueueFull() const;
+
     friend class PulsarWrapper;
 
    private:

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

@@ -473,6 +473,34 @@ class PULSAR_PUBLIC ProducerConfiguration {
      */
     ProducerConfiguration& setProperties(const std::map<std::string, std::string>& properties);
 
+    /**
+     * If message size is higher than allowed max publish-payload size by broker then enableChunking helps
+     * producer to split message into multiple chunks and publish them to broker separately in order. So, it
+     * allows client to successfully publish large size of messages in pulsar.
+     *
+     * Set it true to enable this feature. If so, you must disable batching (see setBatchingEnabled),
+     * otherwise the producer creation will fail.
+     *
+     * There are some other recommendations when it's enabled:
+     * 1. This features is right now only supported for non-shared subscription and persistent-topic.
+     * 2. It's better to reduce setMaxPendingMessages to avoid producer accupying large amount of memory by
+     * buffered messages.
+     * 3. Set message-ttl on the namespace to cleanup chunked messages. Sometimes due to broker-restart or
+     * publish time, producer might fail to publish entire large message. So, consumer will not be able to
+     * consume and ack those messages.
+     *
+     * Default: false
+     *
+     * @param chunkingEnabled whether chunking is enabled
+     * @return the ProducerConfiguration self
+     */
+    ProducerConfiguration& setChunkingEnabled(bool chunkingEnabled);
+
+    /**
+     * The getter associated with setChunkingEnabled().
+     */
+    bool isChunkingEnabled() const;
+
     friend class PulsarWrapper;
 
    private:

pulsar-client-cpp/lib/BatchMessageContainerBase.cc

@@ -74,7 +74,8 @@ Result BatchMessageContainerBase::createOpSendMsgHelper(OpSendMsg& opSendMsg,
         return ResultMessageTooBig;
     }
 
-    opSendMsg.msg_.impl_ = impl;
+    opSendMsg.metadata_ = impl->metadata;
+    opSendMsg.payload_ = impl->payload;
     opSendMsg.sequenceId_ = impl->metadata.sequence_id();
     opSendMsg.producerId_ = producerId_;
     opSendMsg.timeout_ = TimeUtils::now() + milliseconds(producerConfig_.getSendTimeout());

pulsar-client-cpp/lib/ClientConnection.cc

@@ -1366,8 +1366,9 @@ void ClientConnection::sendMessage(const OpSendMsg& opSend) {
 }
 
 void ClientConnection::sendMessageInternal(const OpSendMsg& opSend) {
-    PairSharedBuffer buffer = Commands::newSend(outgoingBuffer_, outgoingCmd_, opSend.producerId_,
-                                                opSend.sequenceId_, getChecksumType(), opSend.msg_);
+    PairSharedBuffer buffer =
+        Commands::newSend(outgoingBuffer_, outgoingCmd_, opSend.producerId_, opSend.sequenceId_,
+                          getChecksumType(), opSend.metadata_, opSend.payload_);
 
     asyncWrite(buffer, customAllocWriteHandler(std::bind(&ClientConnection::handleSendPair,
                                                          shared_from_this(), std::placeholders::_1)));
@@ -1408,8 +1409,9 @@ void ClientConnection::sendPendingCommands() {
             assert(any.type() == typeid(OpSendMsg));
 
             const OpSendMsg& op = boost::any_cast<const OpSendMsg&>(any);
-            PairSharedBuffer buffer = Commands::newSend(outgoingBuffer_, outgoingCmd_, op.producerId_,
-                                                        op.sequenceId_, getChecksumType(), op.msg_);
+            PairSharedBuffer buffer =
+                Commands::newSend(outgoingBuffer_, outgoingCmd_, op.producerId_, op.sequenceId_,
+                                  getChecksumType(), op.metadata_, op.payload_);
 
             asyncWrite(buffer, customAllocWriteHandler(std::bind(&ClientConnection::handleSendPair,
                                                                  shared_from_this(), std::placeholders::_1)));

pulsar-client-cpp/lib/ClientImpl.cc

@@ -29,6 +29,7 @@
 #include <pulsar/ConsoleLoggerFactory.h>
 #include <boost/algorithm/string/predicate.hpp>
 #include <sstream>
+#include <stdexcept>
 #include <lib/HTTPLookupService.h>
 #include <lib/TopicName.h>
 #include <algorithm>
@@ -147,6 +148,9 @@ LookupServicePtr ClientImpl::getLookup() { return lookupServicePtr_; }
 
 void ClientImpl::createProducerAsync(const std::string& topic, ProducerConfiguration conf,
                                      CreateProducerCallback callback) {
+    if (conf.isChunkingEnabled() && conf.getBatchingEnabled()) {
+        throw std::invalid_argument("Batching and chunking of messages can't be enabled together");
+    }
     TopicNamePtr topicName;
     {
         Lock lock(mutex_);
@@ -174,7 +178,7 @@ void ClientImpl::handleCreateProducer(const Result result, const LookupDataResul
             producer = std::make_shared<PartitionedProducerImpl>(shared_from_this(), topicName,
                                                                  partitionMetadata->getPartitions(), conf);
         } else {
-            producer = std::make_shared<ProducerImpl>(shared_from_this(), topicName->toString(), conf);
+            producer = std::make_shared<ProducerImpl>(shared_from_this(), *topicName, conf);
         }
         producer->getProducerCreatedFuture().addListener(
             std::bind(&ClientImpl::handleProducerCreated, shared_from_this(), std::placeholders::_1,

pulsar-client-cpp/lib/Commands.cc

@@ -141,17 +141,18 @@ SharedBuffer Commands::newConsumerStats(uint64_t consumerId, uint64_t requestId)
 }
 
 PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint64_t producerId,
-                                   uint64_t sequenceId, ChecksumType checksumType, const Message& msg) {
-    const proto::MessageMetadata& metadata = msg.impl_->metadata;
-    SharedBuffer& payload = msg.impl_->payload;
-
+                                   uint64_t sequenceId, ChecksumType checksumType,
+                                   const proto::MessageMetadata& metadata, const SharedBuffer& payload) {
     cmd.set_type(BaseCommand::SEND);
     CommandSend* send = cmd.mutable_send();
     send->set_producer_id(producerId);
     send->set_sequence_id(sequenceId);
     if (metadata.has_num_messages_in_batch()) {
         send->set_num_messages(metadata.num_messages_in_batch());
     }
+    if (metadata.has_chunk_id()) {
+        send->set_is_chunk(true);
+    }
 
     // / Wire format
     // [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]
@@ -199,7 +200,8 @@ PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint
         int metadataStartIndex = checksumReaderIndex + checksumSize;
         uint32_t metadataChecksum =
             computeChecksum(0, headers.data() + metadataStartIndex, (writeIndex - metadataStartIndex));
-        uint32_t computedChecksum = computeChecksum(metadataChecksum, payload.data(), payload.writerIndex());
+        uint32_t computedChecksum =
+            computeChecksum(metadataChecksum, payload.data(), payload.readableBytes());
         // set computed checksum
         headers.setWriterIndex(checksumReaderIndex);
         headers.writeUnsignedInt(computedChecksum);

pulsar-client-cpp/lib/Commands.h

@@ -80,7 +80,8 @@ class Commands {
                                   const std::string& listenerName);
 
     static PairSharedBuffer newSend(SharedBuffer& headers, proto::BaseCommand& cmd, uint64_t producerId,
-                                    uint64_t sequenceId, ChecksumType checksumType, const Message& msg);
+                                    uint64_t sequenceId, ChecksumType checksumType,
+                                    const proto::MessageMetadata& metadata, const SharedBuffer& payload);
 
     static SharedBuffer newSubscribe(const std::string& topic, const std::string& subscription,
                                      uint64_t consumerId, uint64_t requestId,

pulsar-client-cpp/lib/ConsumerConfiguration.cc

@@ -231,4 +231,21 @@ ConsumerConfiguration& ConsumerConfiguration::setKeySharedPolicy(KeySharedPolicy
 
 KeySharedPolicy ConsumerConfiguration::getKeySharedPolicy() const { return impl_->keySharedPolicy; }
 
+ConsumerConfiguration& ConsumerConfiguration::setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage) {
+    impl_->maxPendingChunkedMessage = maxPendingChunkedMessage;
+    return *this;
+}
+
+size_t ConsumerConfiguration::getMaxPendingChunkedMessage() const { return impl_->maxPendingChunkedMessage; }
+
+ConsumerConfiguration& ConsumerConfiguration::setAutoOldestChunkedMessageOnQueueFull(
+    bool autoAckOldestChunkedMessageOnQueueFull) {
+    impl_->autoAckOldestChunkedMessageOnQueueFull = autoAckOldestChunkedMessageOnQueueFull;
+    return *this;
+}
+
+bool ConsumerConfiguration::isAutoOldestChunkedMessageOnQueueFull() const {
+    return impl_->autoAckOldestChunkedMessageOnQueueFull;
+}
+
 }  // namespace pulsar

pulsar-client-cpp/lib/ConsumerConfigurationImpl.h

@@ -50,6 +50,8 @@ struct ConsumerConfigurationImpl {
     std::map<std::string, std::string> properties;
     int priorityLevel{0};
     KeySharedPolicy keySharedPolicy;
+    size_t maxPendingChunkedMessage{100};
+    bool autoAckOldestChunkedMessageOnQueueFull{false};
 };
 }  // namespace pulsar
 #endif /* LIB_CONSUMERCONFIGURATIONIMPL_H_ */