diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h
index 201eba38a80d5..2cdbf4707061e 100644
--- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h
+++ b/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:
diff --git a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h
index 5c2792aadeaa3..7c278dd6e9124 100644
--- a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h
+++ b/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:
diff --git a/pulsar-client-cpp/lib/BatchMessageContainerBase.cc b/pulsar-client-cpp/lib/BatchMessageContainerBase.cc
index 1201707e0645e..e9e6b987a107e 100644
--- a/pulsar-client-cpp/lib/BatchMessageContainerBase.cc
+++ b/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());
diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc
index 3ad6f4062f4b9..79bc1d74aa967 100644
--- a/pulsar-client-cpp/lib/ClientConnection.cc
+++ b/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)));
diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc
index 8cdaacc8b2523..0b07b6e9f2d5c 100644
--- a/pulsar-client-cpp/lib/ClientImpl.cc
+++ b/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,
diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc
index 1094efb8eee35..472c15bed8a75 100644
--- a/pulsar-client-cpp/lib/Commands.cc
+++ b/pulsar-client-cpp/lib/Commands.cc
@@ -141,10 +141,8 @@ 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);
@@ -152,6 +150,9 @@ PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint
     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);
diff --git a/pulsar-client-cpp/lib/Commands.h b/pulsar-client-cpp/lib/Commands.h
index e72057182ff2c..bab2211f7fa95 100644
--- a/pulsar-client-cpp/lib/Commands.h
+++ b/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,
diff --git a/pulsar-client-cpp/lib/ConsumerConfiguration.cc b/pulsar-client-cpp/lib/ConsumerConfiguration.cc
index b01e4e595b6d7..d13cb0e0c86b2 100644
--- a/pulsar-client-cpp/lib/ConsumerConfiguration.cc
+++ b/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
diff --git a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h
index 75f65a7eb8eab..9c2a4615fef43 100644
--- a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h
+++ b/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_ */
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc
index 77c0fa9d52a94..4d4a135dbec69 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -63,7 +63,9 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
       negativeAcksTracker_(client, *this, conf),
       ackGroupingTrackerPtr_(std::make_shared<AckGroupingTracker>()),
       readCompacted_(conf.isReadCompacted()),
-      lastMessageInBroker_(Optional<MessageId>::of(MessageId())) {
+      lastMessageInBroker_(Optional<MessageId>::of(MessageId())),
+      maxPendingChunkedMessage_(conf.getMaxPendingChunkedMessage()),
+      autoAckOldestChunkedMessageOnQueueFull_(conf.isAutoOldestChunkedMessageOnQueueFull()) {
     std::stringstream consumerStrStream;
     consumerStrStream << "[" << topic_ << ", " << subscription_ << ", " << consumerId_ << "] ";
     consumerStr_ = consumerStrStream.str();
@@ -308,6 +310,80 @@ void ConsumerImpl::handleUnsubscribe(Result result, ResultCallback callback) {
     callback(result);
 }
 
+Optional<SharedBuffer> ConsumerImpl::processMessageChunk(const SharedBuffer& payload,
+                                                         const proto::MessageMetadata& metadata,
+                                                         const MessageId& messageId,
+                                                         const proto::MessageIdData& messageIdData,
+                                                         const ClientConnectionPtr& cnx) {
+    const auto chunkId = metadata.chunk_id();
+    const auto uuid = metadata.uuid();
+    LOG_DEBUG("Process message chunk (chunkId: " << chunkId << ", uuid: " << uuid
+                                                 << ", messageId: " << messageId << ") of "
+                                                 << payload.readableBytes() << " bytes");
+
+    Lock lock(chunkProcessMutex_);
+    auto it = chunkedMessageCache_.find(uuid);
+
+    if (chunkId == 0) {
+        if (it == chunkedMessageCache_.end()) {
+            it = chunkedMessageCache_.putIfAbsent(
+                uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+        }
+        if (maxPendingChunkedMessage_ > 0 && chunkedMessageCache_.size() >= maxPendingChunkedMessage_) {
+            chunkedMessageCache_.removeOldestValues(
+                chunkedMessageCache_.size() - maxPendingChunkedMessage_ + 1,
+                [this, messageId](const std::string& uuid, const ChunkedMessageCtx& ctx) {
+                    if (autoAckOldestChunkedMessageOnQueueFull_) {
+                        doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) {
+                            if (result != ResultOk) {
+                                LOG_WARN("Failed to acknowledge discarded chunk, uuid: "
+                                         << uuid << ", messageId: " << messageId);
+                            }
+                        });
+                    } else {
+                        trackMessage(messageId);
+                    }
+                });
+            it = chunkedMessageCache_.putIfAbsent(
+                uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()});
+        }
+    }
+
+    auto& chunkedMsgCtx = it->second;
+    if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) {
+        if (it == chunkedMessageCache_.end()) {
+            LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId
+                                                           << ", messageId: " << messageId << ")");
+        } else {
+            LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: "
+                      << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")");
+            chunkedMessageCache_.remove(uuid);
+        }
+        lock.unlock();
+        increaseAvailablePermits(cnx);
+        trackMessage(messageId);
+        return Optional<SharedBuffer>::empty();
+    }
+
+    chunkedMsgCtx.appendChunk(messageId, payload);
+    if (!chunkedMsgCtx.isCompleted()) {
+        lock.unlock();
+        increaseAvailablePermits(cnx);
+        return Optional<SharedBuffer>::empty();
+    }
+
+    LOG_DEBUG("Chunked message completed chunkId: " << chunkId << ", ChunkedMessageCtx: " << chunkedMsgCtx
+                                                    << ", sequenceId: " << metadata.sequence_id());
+
+    auto wholePayload = chunkedMsgCtx.getBuffer();
+    chunkedMessageCache_.remove(uuid);
+    if (uncompressMessageIfNeeded(cnx, messageIdData, metadata, wholePayload, false)) {
+        return Optional<SharedBuffer>::of(wholePayload);
+    } else {
+        return Optional<SharedBuffer>::empty();
+    }
+}
+
 void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
                                    bool& isChecksumValid, proto::MessageMetadata& metadata,
                                    SharedBuffer& payload) {
@@ -318,17 +394,39 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
         return;
     }
 
-    if (!uncompressMessageIfNeeded(cnx, msg, metadata, payload)) {
-        // Message was discarded on decompression error
-        return;
-    }
-
     if (!isChecksumValid) {
         // Message discarded for checksum error
         discardCorruptedMessage(cnx, msg.message_id(), proto::CommandAck::ChecksumMismatch);
         return;
     }
 
+    const bool isMessageDecryptable =
+        metadata.encryption_keys_size() <= 0 || config_.getCryptoKeyReader().get() ||
+        config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::CONSUME;
+
+    const bool isChunkedMessage = metadata.num_chunks_from_msg() > 1 &&
+                                  config_.getConsumerType() != ConsumerType::ConsumerShared &&
+                                  config_.getConsumerType() != ConsumerType::ConsumerKeyShared;
+    if (isMessageDecryptable && !isChunkedMessage) {
+        if (!uncompressMessageIfNeeded(cnx, msg.message_id(), metadata, payload, true)) {
+            // Message was discarded on decompression error
+            return;
+        }
+    }
+
+    // Only a non-batched messages can be a chunk
+    if (!metadata.has_num_messages_in_batch() && isChunkedMessage) {
+        const auto& messageIdData = msg.message_id();
+        MessageId messageId(messageIdData.partition(), messageIdData.ledgerid(), messageIdData.entryid(),
+                            messageIdData.batch_index());
+        auto optionalPayload = processMessageChunk(payload, metadata, messageId, messageIdData, cnx);
+        if (optionalPayload.is_present()) {
+            payload = optionalPayload.value();
+        } else {
+            return;
+        }
+    }
+
     Message m(msg, metadata, payload, partitionIndex_);
     m.impl_->cnx_ = cnx.get();
     m.impl_->setTopicName(topic_);
@@ -528,8 +626,10 @@ bool ConsumerImpl::decryptMessageIfNeeded(const ClientConnectionPtr& cnx, const
     return false;
 }
 
-bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
-                                             const proto::MessageMetadata& metadata, SharedBuffer& payload) {
+bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx,
+                                             const proto::MessageIdData& messageIdData,
+                                             const proto::MessageMetadata& metadata, SharedBuffer& payload,
+                                             bool checkMaxMessageSize) {
     if (!metadata.has_compression()) {
         return true;
     }
@@ -539,12 +639,11 @@ bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, con
     uint32_t uncompressedSize = metadata.uncompressed_size();
     uint32_t payloadSize = payload.readableBytes();
     if (cnx) {
-        if (payloadSize > ClientConnection::getMaxMessageSize()) {
+        if (checkMaxMessageSize && payloadSize > ClientConnection::getMaxMessageSize()) {
             // Uncompressed size is itself corrupted since it cannot be bigger than the MaxMessageSize
             LOG_ERROR(getName() << "Got corrupted payload message size " << payloadSize  //
-                                << " at  " << msg.message_id().ledgerid() << ":"
-                                << msg.message_id().entryid());
-            discardCorruptedMessage(cnx, msg.message_id(), proto::CommandAck::UncompressedSizeCorruption);
+                                << " at  " << messageIdData.ledgerid() << ":" << messageIdData.entryid());
+            discardCorruptedMessage(cnx, messageIdData, proto::CommandAck::UncompressedSizeCorruption);
             return false;
         }
     } else {
@@ -554,8 +653,8 @@ bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, con
 
     if (!CompressionCodecProvider::getCodec(compressionType).decode(payload, uncompressedSize, payload)) {
         LOG_ERROR(getName() << "Failed to decompress message with " << uncompressedSize  //
-                            << " at  " << msg.message_id().ledgerid() << ":" << msg.message_id().entryid());
-        discardCorruptedMessage(cnx, msg.message_id(), proto::CommandAck::DecompressionError);
+                            << " at  " << messageIdData.ledgerid() << ":" << messageIdData.entryid());
+        discardCorruptedMessage(cnx, messageIdData, proto::CommandAck::DecompressionError);
         return false;
     }
 
@@ -584,7 +683,7 @@ void ConsumerImpl::internalListener() {
         // This will only happen when the connection got reset and we cleared the queue
         return;
     }
-    trackMessage(msg);
+    trackMessage(msg.getMessageId());
     try {
         consumerStatsBasePtr_->receivedMessage(msg, ResultOk);
         lastDequedMessage_ = Optional<MessageId>::of(msg.getMessageId());
@@ -732,7 +831,7 @@ void ConsumerImpl::messageProcessed(Message& msg, bool track) {
 
     increaseAvailablePermits(currentCnx);
     if (track) {
-        trackMessage(msg);
+        trackMessage(msg.getMessageId());
     }
 }
 
@@ -1240,11 +1339,11 @@ void ConsumerImpl::setNegativeAcknowledgeEnabledForTesting(bool enabled) {
     negativeAcksTracker_.setEnabledForTesting(enabled);
 }
 
-void ConsumerImpl::trackMessage(const Message& msg) {
+void ConsumerImpl::trackMessage(const MessageId& messageId) {
     if (hasParent_) {
-        unAckedMessageTrackerPtr_->remove(msg.getMessageId());
+        unAckedMessageTrackerPtr_->remove(messageId);
     } else {
-        unAckedMessageTrackerPtr_->add(msg.getMessageId());
+        unAckedMessageTrackerPtr_->add(messageId);
     }
 }
 
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h
index 0754a89fdf5fc..2bdb82fc7630a 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.h
+++ b/pulsar-client-cpp/lib/ConsumerImpl.h
@@ -40,6 +40,7 @@
 #include "BatchAcknowledgementTracker.h"
 #include <limits>
 #include <lib/BrokerConsumerStatsImpl.h>
+#include <lib/MapCache.h>
 #include <lib/stats/ConsumerStatsImpl.h>
 #include <lib/stats/ConsumerStatsDisabled.h>
 #include <queue>
@@ -157,8 +158,9 @@ class ConsumerImpl : public ConsumerImplBase,
 
    private:
     bool waitingForZeroQueueSizeMessage;
-    bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
-                                   const proto::MessageMetadata& metadata, SharedBuffer& payload);
+    bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData,
+                                   const proto::MessageMetadata& metadata, SharedBuffer& payload,
+                                   bool checkMaxMessageSize);
     void discardCorruptedMessage(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageId,
                                  proto::CommandAck::ValidationError validationError);
     void increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int delta = 1);
@@ -177,7 +179,7 @@ class ConsumerImpl : public ConsumerImplBase,
     void notifyPendingReceivedCallback(Result result, Message& message, const ReceiveCallback& callback);
     void failPendingReceiveCallback();
     void setNegativeAcknowledgeEnabledForTesting(bool enabled) override;
-    void trackMessage(const Message& msg);
+    void trackMessage(const MessageId& messageId);
 
     Optional<MessageId> clearReceiveQueue();
 
@@ -227,6 +229,83 @@ class ConsumerImpl : public ConsumerImplBase,
         return lastMessageInBroker_.is_present() ? lastMessageInBroker_.value() : MessageId::earliest();
     }
 
+    class ChunkedMessageCtx {
+       public:
+        ChunkedMessageCtx() : totalChunks_(0) {}
+        ChunkedMessageCtx(int totalChunks, int totalChunkMessageSize)
+            : totalChunks_(totalChunks), chunkedMsgBuffer_(SharedBuffer::allocate(totalChunkMessageSize)) {
+            chunkedMessageIds_.reserve(totalChunks);
+        }
+
+        ChunkedMessageCtx(const ChunkedMessageCtx&) = delete;
+        // Here we don't use =default to be compatible with GCC 4.8
+        ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept
+            : totalChunks_(rhs.totalChunks_),
+              chunkedMsgBuffer_(std::move(rhs.chunkedMsgBuffer_)),
+              chunkedMessageIds_(std::move(rhs.chunkedMessageIds_)) {}
+
+        bool validateChunkId(int chunkId) const noexcept { return chunkId == numChunks(); }
+
+        void appendChunk(const MessageId& messageId, const SharedBuffer& payload) {
+            chunkedMessageIds_.emplace_back(messageId);
+            chunkedMsgBuffer_.write(payload.data(), payload.readableBytes());
+        }
+
+        bool isCompleted() const noexcept { return totalChunks_ == numChunks(); }
+
+        const SharedBuffer& getBuffer() const noexcept { return chunkedMsgBuffer_; }
+
+        const std::vector<MessageId>& getChunkedMessageIds() const noexcept { return chunkedMessageIds_; }
+
+        friend std::ostream& operator<<(std::ostream& os, const ChunkedMessageCtx& ctx) {
+            return os << "ChunkedMessageCtx " << ctx.chunkedMsgBuffer_.readableBytes() << " of "
+                      << ctx.chunkedMsgBuffer_.writerIndex() << " bytes, " << ctx.numChunks() << " of "
+                      << ctx.totalChunks_ << " chunks";
+        }
+
+       private:
+        const int totalChunks_;
+        SharedBuffer chunkedMsgBuffer_;
+        std::vector<MessageId> chunkedMessageIds_;
+
+        int numChunks() const noexcept { return static_cast<int>(chunkedMessageIds_.size()); }
+    };
+
+    const size_t maxPendingChunkedMessage_;
+    // if queue size is reasonable (most of the time equal to number of producers try to publish messages
+    // concurrently on the topic) then it guards against broken chunked message which was not fully published
+    const bool autoAckOldestChunkedMessageOnQueueFull_;
+
+    // The key is UUID, value is the associated ChunkedMessageCtx of the chunked message.
+    std::unordered_map<std::string, ChunkedMessageCtx> chunkedMessagesMap_;
+    // This list contains all the keys of `chunkedMessagesMap_`, each key is an UUID that identifies a pending
+    // chunked message. Once the number of pending chunked messages exceeds the limit, the oldest UUIDs and
+    // the associated ChunkedMessageCtx will be removed.
+    std::list<std::string> pendingChunkedMessageUuidQueue_;
+
+    // The key is UUID, value is the associated ChunkedMessageCtx of the chunked message.
+    MapCache<std::string, ChunkedMessageCtx> chunkedMessageCache_;
+    mutable std::mutex chunkProcessMutex_;
+
+    /**
+     * Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the
+     * payload and return it.
+     *
+     * @param payload the payload of a chunk
+     * @param metadata the message metadata
+     * @param messageId
+     * @param messageIdData
+     * @param cnx
+     *
+     * @return the concatenated payload if chunks are concatenated into a completed message payload
+     *   successfully, else Optional::empty()
+     */
+    Optional<SharedBuffer> processMessageChunk(const SharedBuffer& payload,
+                                               const proto::MessageMetadata& metadata,
+                                               const MessageId& messageId,
+                                               const proto::MessageIdData& messageIdData,
+                                               const ClientConnectionPtr& cnx);
+
     friend class PulsarFriend;
 
     // these two declared friend to access setNegativeAcknowledgeEnabledForTesting
diff --git a/pulsar-client-cpp/lib/MapCache.h b/pulsar-client-cpp/lib/MapCache.h
new file mode 100644
index 0000000000000..b9a0069eaf44a
--- /dev/null
+++ b/pulsar-client-cpp/lib/MapCache.h
@@ -0,0 +1,104 @@
+/**
+ * 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 <deque>
+#include <functional>
+#include <unordered_map>
+#include <vector>
+
+namespace pulsar {
+
+// A map cache that supports removing the first N oldest entries from the map.
+// Value must be moveable and have the default constructor.
+template <typename Key, typename Value>
+class MapCache {
+    std::unordered_map<Key, Value> map_;
+    std::deque<Key> keys_;
+
+   public:
+    using const_iterator = typename decltype(map_)::const_iterator;
+    using iterator = typename decltype(map_)::iterator;
+
+    MapCache() = default;
+    // Here we don't use =default to be compatible with GCC 4.8
+    MapCache(MapCache&& rhs) noexcept : map_(std::move(rhs.map_)), keys_(std::move(rhs.keys_)) {}
+
+    size_t size() const noexcept { return map_.size(); }
+
+    const_iterator find(const Key& key) const { return map_.find(key); }
+    iterator find(const Key& key) { return map_.find(key); }
+
+    const_iterator end() const noexcept { return map_.end(); }
+    iterator end() noexcept { return map_.end(); }
+
+    iterator putIfAbsent(const Key& key, Value&& value) {
+        auto it = map_.find(key);
+        if (it == map_.end()) {
+            keys_.push_back(key);
+            return map_.emplace(key, std::move(value)).first;
+        } else {
+            return end();
+        }
+    }
+
+    void removeOldestValues(size_t numToRemove,
+                            const std::function<void(const Key&, const Value&)>& callback) {
+        for (size_t i = 0; !keys_.empty() && i < numToRemove; i++) {
+            const auto key = keys_.front();
+            auto it = map_.find(key);
+            if (it != map_.end()) {
+                if (callback) {
+                    callback(it->first, it->second);
+                }
+                map_.erase(it);
+            }
+            keys_.pop_front();
+        }
+    }
+
+    void remove(const Key& key) {
+        auto it = map_.find(key);
+        if (it != map_.end()) {
+            removeKeyFromKeys(key);
+            map_.erase(it);
+        }
+    }
+
+    // Following methods are only used for tests
+    std::vector<Key> getKeys() const {
+        std::vector<Key> keys;
+        for (auto key : keys_) {
+            keys.emplace_back(key);
+        }
+        return keys;
+    }
+
+   private:
+    void removeKeyFromKeys(const Key& key) {
+        for (auto it = keys_.begin(); it != keys_.end(); ++it) {
+            if (*it == key) {
+                keys_.erase(it);
+                break;
+            }
+        }
+    }
+};
+
+}  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/MemoryLimitController.cc b/pulsar-client-cpp/lib/MemoryLimitController.cc
index 81578c9e3f4fb..4a23f8b1a1d0c 100644
--- a/pulsar-client-cpp/lib/MemoryLimitController.cc
+++ b/pulsar-client-cpp/lib/MemoryLimitController.cc
@@ -25,6 +25,10 @@ MemoryLimitController::MemoryLimitController(uint64_t memoryLimit)
     : memoryLimit_(memoryLimit), currentUsage_(0), mutex_(), condition_() {}
 
 bool MemoryLimitController::tryReserveMemory(uint64_t size) {
+    // Avoid CAS operation when size is 0
+    if (size == 0) {
+        return true;
+    }
     while (true) {
         uint64_t current = currentUsage_;
         uint64_t newUsage = current + size;
@@ -66,4 +70,4 @@ void MemoryLimitController::releaseMemory(uint64_t size) {
 
 uint64_t MemoryLimitController::currentUsage() const { return currentUsage_; }
 
-}  // namespace pulsar
\ No newline at end of file
+}  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/OpSendMsg.h b/pulsar-client-cpp/lib/OpSendMsg.h
index 70b880cf5179c..365301be4ea95 100644
--- a/pulsar-client-cpp/lib/OpSendMsg.h
+++ b/pulsar-client-cpp/lib/OpSendMsg.h
@@ -29,7 +29,8 @@
 namespace pulsar {
 
 struct OpSendMsg {
-    Message msg_;
+    proto::MessageMetadata metadata_;
+    SharedBuffer payload_;
     SendCallback sendCallback_;
     uint64_t producerId_;
     uint64_t sequenceId_;
@@ -39,15 +40,24 @@ struct OpSendMsg {
 
     OpSendMsg() = default;
 
-    OpSendMsg(const Message& msg, const SendCallback& sendCallback, uint64_t producerId, uint64_t sequenceId,
-              int sendTimeoutMs, uint32_t messagesCount, uint64_t messagesSize)
-        : msg_(msg),
+    OpSendMsg(const proto::MessageMetadata& metadata, const SharedBuffer& payload,
+              const SendCallback& sendCallback, uint64_t producerId, uint64_t sequenceId, int sendTimeoutMs,
+              uint32_t messagesCount, uint64_t messagesSize)
+        : metadata_(metadata),  // the copy happens here because OpSendMsg of chunks are constructed with the
+                                // a shared metadata object
+          payload_(payload),
           sendCallback_(sendCallback),
           producerId_(producerId),
           sequenceId_(sequenceId),
           timeout_(TimeUtils::now() + milliseconds(sendTimeoutMs)),
           messagesCount_(messagesCount),
           messagesSize_(messagesSize) {}
+
+    void complete(Result result, const MessageId& messageId) const {
+        if (sendCallback_) {
+            sendCallback_(result, messageId);
+        }
+    }
 };
 
 }  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc
index bdd23ed6c91c2..4f8be3b92becc 100644
--- a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc
+++ b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc
@@ -86,8 +86,7 @@ unsigned int PartitionedProducerImpl::getNumPartitionsWithLock() 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);
+    auto producer = std::make_shared<ProducerImpl>(client_, *topicName_, conf_, partition);
 
     if (lazy) {
         createLazyPartitionProducer(partition);
@@ -97,7 +96,7 @@ ProducerImplPtr PartitionedProducerImpl::newInternalProducer(unsigned int partit
                       const_cast<PartitionedProducerImpl*>(this)->shared_from_this(), _1, _2, partition));
     }
 
-    LOG_DEBUG("Creating Producer for single Partition - " << topicPartitionName);
+    LOG_DEBUG("Creating Producer for single Partition - " << topicName_ << "-partition-" << partition);
     return producer;
 }
 
diff --git a/pulsar-client-cpp/lib/ProducerConfiguration.cc b/pulsar-client-cpp/lib/ProducerConfiguration.cc
index 3e027eeb1975f..0ee38ca92e025 100644
--- a/pulsar-client-cpp/lib/ProducerConfiguration.cc
+++ b/pulsar-client-cpp/lib/ProducerConfiguration.cc
@@ -251,4 +251,11 @@ ProducerConfiguration& ProducerConfiguration::setProperties(
     return *this;
 }
 
+ProducerConfiguration& ProducerConfiguration::setChunkingEnabled(bool chunkingEnabled) {
+    impl_->chunkingEnabled = chunkingEnabled;
+    return *this;
+}
+
+bool ProducerConfiguration::isChunkingEnabled() const { return impl_->chunkingEnabled; }
+
 }  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/ProducerConfigurationImpl.h b/pulsar-client-cpp/lib/ProducerConfigurationImpl.h
index a41b2507ea43c..2ac1ebaa5df1a 100644
--- a/pulsar-client-cpp/lib/ProducerConfigurationImpl.h
+++ b/pulsar-client-cpp/lib/ProducerConfigurationImpl.h
@@ -48,6 +48,7 @@ struct ProducerConfigurationImpl {
     std::set<std::string> encryptionKeys;
     ProducerCryptoFailureAction cryptoFailureAction{ProducerCryptoFailureAction::FAIL};
     std::map<std::string, std::string> properties;
+    bool chunkingEnabled{false};
 };
 }  // namespace pulsar
 
diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc
index f81e205475de4..87f8552ee4516 100644
--- a/pulsar-client-cpp/lib/ProducerImpl.cc
+++ b/pulsar-client-cpp/lib/ProducerImpl.cc
@@ -37,15 +37,15 @@ struct ProducerImpl::PendingCallbacks {
 
     void complete(Result result) {
         for (const auto& opSendMsg : opSendMsgs) {
-            opSendMsg.sendCallback_(result, opSendMsg.msg_.getMessageId());
+            opSendMsg.complete(result, {});
         }
     }
 };
 
-ProducerImpl::ProducerImpl(ClientImplPtr client, const std::string& topic, const ProducerConfiguration& conf,
-                           int32_t partition)
+ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName,
+                           const ProducerConfiguration& conf, int32_t partition)
     : HandlerBase(
-          client, topic,
+          client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition),
           Backoff(milliseconds(100), seconds(60), milliseconds(std::max(100, conf.getSendTimeout() - 100)))),
       conf_(conf),
       semaphore_(),
@@ -57,7 +57,8 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const std::string& topic, const
       producerId_(client->newProducerId()),
       msgSequenceGenerator_(0),
       dataKeyGenIntervalSec_(4 * 60 * 60),
-      memoryLimitController_(client->getMemoryLimitController()) {
+      memoryLimitController_(client->getMemoryLimitController()),
+      chunkingEnabled_(conf_.isChunkingEnabled() && topicName.isPersistent() && !conf_.getBatchingEnabled()) {
     LOG_DEBUG("ProducerName - " << producerName_ << " Created producer on topic " << topic_
                                 << " id: " << producerId_);
 
@@ -323,14 +324,6 @@ void ProducerImpl::setMessageMetadata(const Message& msg, const uint64_t& sequen
     }
 }
 
-void ProducerImpl::statsCallBackHandler(Result res, const MessageId& msgId, SendCallback callback,
-                                        boost::posix_time::ptime publishTime) {
-    producerStatsBasePtr_->messageReceived(res, publishTime);
-    if (callback) {
-        callback(res, msgId);
-    }
-}
-
 void ProducerImpl::flushAsync(FlushCallback callback) {
     if (batchMessageContainer_) {
         Lock lock(mutex_);
@@ -358,44 +351,61 @@ void ProducerImpl::triggerFlush() {
     }
 }
 
+bool ProducerImpl::isValidProducerState(const SendCallback& callback) const {
+    Lock lock(mutex_);
+    const auto state = state_;
+    lock.unlock();
+    switch (state) {
+        case HandlerBase::Ready:
+            // OK
+        case HandlerBase::Pending:
+            // We are OK to queue the messages on the client, it will be sent to the broker once we get the
+            // connection
+            return true;
+        case HandlerBase::Closing:
+        case HandlerBase::Closed:
+            callback(ResultAlreadyClosed, {});
+            return false;
+        case HandlerBase::NotStarted:
+        case HandlerBase::Failed:
+        default:
+            callback(ResultNotConnected, {});
+            return false;
+    }
+}
+
+bool ProducerImpl::canAddToBatch(const Message& msg) const {
+    // If a message has a delayed delivery time, we'll always send it individually
+    return batchMessageContainer_.get() && !msg.impl_->metadata.has_deliver_at_time();
+}
+
+static SharedBuffer applyCompression(const SharedBuffer& uncompressedPayload,
+                                     CompressionType compressionType) {
+    return CompressionCodecProvider::getCodec(compressionType).encode(uncompressedPayload);
+}
+
 void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) {
     producerStatsBasePtr_->messageSent(msg);
-    SendCallback cb =
-        std::bind(&ProducerImpl::statsCallBackHandler, shared_from_this(), std::placeholders::_1,
-                  std::placeholders::_2, callback, boost::posix_time::microsec_clock::universal_time());
-
-    // Compress the payload if required
-    SharedBuffer& payload = msg.impl_->payload;
 
-    uint32_t uncompressedSize = payload.readableBytes();
-    uint32_t payloadSize = uncompressedSize;
-    ClientConnectionPtr cnx = getCnx().lock();
-    if (!batchMessageContainer_) {
-        // If batching is enabled we compress all the payloads together before sending the batch
-        payload = CompressionCodecProvider::getCodec(conf_.getCompressionType()).encode(payload);
-        payloadSize = payload.readableBytes();
-
-        // Encrypt the payload if enabled
-        SharedBuffer encryptedPayload;
-        if (!encryptMessage(msg.impl_->metadata, payload, encryptedPayload)) {
-            cb(ResultCryptoError, msg.getMessageId());
-            return;
+    const auto now = boost::posix_time::microsec_clock::universal_time();
+    auto self = shared_from_this();
+    sendAsyncWithStatsUpdate(msg, [this, self, now, callback](Result result, const MessageId& messageId) {
+        producerStatsBasePtr_->messageReceived(result, now);
+        if (callback) {
+            callback(result, messageId);
         }
-        payload = encryptedPayload;
+    });
+}
 
-        if (payloadSize > ClientConnection::getMaxMessageSize()) {
-            LOG_DEBUG(getName() << " - compressed Message payload size" << payloadSize << "cannot exceed "
-                                << ClientConnection::getMaxMessageSize() << " bytes");
-            cb(ResultMessageTooBig, msg.getMessageId());
-            return;
-        }
+void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallback& callback) {
+    if (!isValidProducerState(callback)) {
+        return;
     }
 
-    // Reserve a spot in the messages queue before acquiring the ProducerImpl
-    // mutex. When the queue is full, this call will block until a spot is
-    // available.
-    Result res = canEnqueueRequest(payloadSize);
-    if (res != ResultOk) {
+    const auto& uncompressedPayload = msg.impl_->payload;
+    const uint32_t uncompressedSize = uncompressedPayload.readableBytes();
+    const auto result = canEnqueueRequest(uncompressedSize);
+    if (result != ResultOk) {
         // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout
         if (batchMessageContainer_) {
             LOG_DEBUG(getName() << " - sending batch message immediately");
@@ -405,43 +415,64 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) {
             failures.complete();
         }
 
-        cb(res, msg.getMessageId());
+        callback(result, {});
         return;
     }
 
-    Lock lock(mutex_);
-    // producers may be lazily starting and be in the pending state
-    if (state_ != Ready && state_ != Pending) {
-        lock.unlock();
-        releaseSemaphore(payloadSize);
-        cb(ResultAlreadyClosed, msg.getMessageId());
+    // We have already reserved a spot, so if we need to early return for failed result, we should release the
+    // semaphore and memory first.
+    const auto handleFailedResult = [this, uncompressedSize, callback](Result result) {
+        releaseSemaphore(uncompressedSize);  // it releases the memory as well
+        callback(result, {});
+    };
+
+    const bool compressed = !canAddToBatch(msg);
+    const auto payload =
+        compressed ? applyCompression(uncompressedPayload, conf_.getCompressionType()) : uncompressedPayload;
+    const auto compressedSize = static_cast<uint32_t>(payload.readableBytes());
+    const auto maxMessageSize = static_cast<uint32_t>(ClientConnection::getMaxMessageSize());
+
+    if (compressed && compressedSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) {
+        LOG_WARN(getName() << " - compressed Message payload size " << payload.readableBytes()
+                           << " cannot exceed " << ClientConnection::getMaxMessageSize()
+                           << " bytes unless chunking is enabled");
+        handleFailedResult(ResultMessageTooBig);
         return;
     }
 
-    if (msg.impl_->metadata.has_producer_name()) {
-        // Message had already been sent before
-        lock.unlock();
-        releaseSemaphore(payloadSize);
-        cb(ResultInvalidMessage, msg.getMessageId());
+    auto& msgMetadata = msg.impl_->metadata;
+    if (!msgMetadata.has_replicated_from() && msgMetadata.has_producer_name()) {
+        handleFailedResult(ResultInvalidMessage);
         return;
     }
 
+    const int totalChunks =
+        canAddToBatch(msg) ? 1 : getNumOfChunks(compressedSize, ClientConnection::getMaxMessageSize());
+    // Each chunk should be sent individually, so try to acquire extra permits for chunks.
+    for (int i = 0; i < (totalChunks - 1); i++) {
+        const auto result = canEnqueueRequest(0);  // size is 0 because the memory has already reserved
+        if (result != ResultOk) {
+            handleFailedResult(result);
+            return;
+        }
+    }
+
+    Lock lock(mutex_);
     uint64_t sequenceId;
-    if (!msg.impl_->metadata.has_sequence_id()) {
+    if (!msgMetadata.has_sequence_id()) {
         sequenceId = msgSequenceGenerator_++;
     } else {
-        sequenceId = msg.impl_->metadata.sequence_id();
+        sequenceId = msgMetadata.sequence_id();
     }
     setMessageMetadata(msg, sequenceId, uncompressedSize);
 
-    // If we reach this point then you have a reserved spot on the queue
-    if (batchMessageContainer_ && !msg.impl_->metadata.has_deliver_at_time()) {
+    if (canAddToBatch(msg)) {
         // Batching is enabled and the message is not delayed
         if (!batchMessageContainer_->hasEnoughSpace(msg)) {
             batchMessageAndSend().complete();
         }
         bool isFirstMessage = batchMessageContainer_->isFirstMessageToAdd(msg);
-        bool isFull = batchMessageContainer_->add(msg, cb);
+        bool isFull = batchMessageContainer_->add(msg, callback);
         if (isFirstMessage) {
             batchTimer_->expires_from_now(
                 boost::posix_time::milliseconds(conf_.getBatchingMaxPublishDelayMs()));
@@ -455,10 +486,42 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) {
             failures.complete();
         }
     } else {
-        sendMessage(OpSendMsg{msg, cb, producerId_, sequenceId, conf_.getSendTimeout(), 1, payloadSize});
+        const bool sendChunks = (totalChunks > 1);
+        if (sendChunks) {
+            msgMetadata.set_uuid(producerName_ + "-" + std::to_string(sequenceId));
+            msgMetadata.set_num_chunks_from_msg(totalChunks);
+            msgMetadata.set_total_chunk_msg_size(compressedSize);
+        }
+
+        int beginIndex = 0;
+        for (int chunkId = 0; chunkId < totalChunks; chunkId++) {
+            if (sendChunks) {
+                msgMetadata.set_chunk_id(chunkId);
+            }
+            const uint32_t endIndex = std::min(compressedSize, beginIndex + maxMessageSize);
+            auto chunkedPayload = payload.slice(beginIndex, endIndex - beginIndex);
+            beginIndex = endIndex;
+
+            SharedBuffer encryptedPayload;
+            if (!encryptMessage(msgMetadata, chunkedPayload, encryptedPayload)) {
+                handleFailedResult(ResultCryptoError);
+                return;
+            }
+
+            sendMessage(OpSendMsg{msgMetadata, encryptedPayload,
+                                  (chunkId == totalChunks - 1) ? callback : nullptr, producerId_, sequenceId,
+                                  conf_.getSendTimeout(), 1, uncompressedSize});
+        }
     }
 }
 
+int ProducerImpl::getNumOfChunks(uint32_t size, uint32_t maxMessageSize) {
+    if (size >= maxMessageSize && maxMessageSize != 0) {
+        return size / maxMessageSize + ((size % maxMessageSize == 0) ? 0 : 1);
+    }
+    return 1;
+}
+
 Result ProducerImpl::canEnqueueRequest(uint32_t payloadSize) {
     if (conf_.getBlockIfQueueFull()) {
         if (semaphore_) {
@@ -523,7 +586,7 @@ PendingFailures ProducerImpl::batchMessageAndSend(const FlushCallback& flushCall
                 // we need to release the spot manually
                 LOG_ERROR("batchMessageAndSend | Failed to createOpSendMsg: " << result);
                 releaseSemaphoreForSendOp(opSendMsg);
-                failures.add(std::bind(opSendMsg.sendCallback_, result, MessageId{}));
+                failures.add([opSendMsg, result] { opSendMsg.complete(result, {}); });
             }
         } else if (numBatches > 1) {
             std::vector<OpSendMsg> opSendMsgs;
@@ -537,7 +600,9 @@ PendingFailures ProducerImpl::batchMessageAndSend(const FlushCallback& flushCall
                     LOG_ERROR("batchMessageAndSend | Failed to createOpSendMsgs[" << i
                                                                                   << "]: " << results[i]);
                     releaseSemaphoreForSendOp(opSendMsgs[i]);
-                    failures.add(std::bind(opSendMsgs[i].sendCallback_, results[i], MessageId{}));
+                    const auto& opSendMsg = opSendMsgs[i];
+                    const auto result = results[i];
+                    failures.add([opSendMsg, result] { opSendMsg.complete(result, {}); });
                 }
             }
         }  // else numBatches is 0, do nothing
@@ -551,7 +616,7 @@ PendingFailures ProducerImpl::batchMessageAndSend(const FlushCallback& flushCall
 // a. we have a reserved spot on the queue
 // b. call this function after acquiring the ProducerImpl mutex_
 void ProducerImpl::sendMessage(const OpSendMsg& op) {
-    const auto sequenceId = op.msg_.impl_->metadata.sequence_id();
+    const auto sequenceId = op.metadata_.sequence_id();
     LOG_DEBUG("Inserting data to pendingMessagesQueue_");
     pendingMessagesQueue_.push_back(op);
 
@@ -748,13 +813,11 @@ bool ProducerImpl::removeCorruptMessage(uint64_t sequenceId) {
         LOG_DEBUG(getName() << "Remove corrupt message from queue " << sequenceId);
         pendingMessagesQueue_.pop_front();
         lock.unlock();
-        if (op.sendCallback_) {
+        try {
             // to protect from client callback exception
-            try {
-                op.sendCallback_(ResultChecksumError, op.msg_.getMessageId());
-            } catch (const std::exception& e) {
-                LOG_ERROR(getName() << "Exception thrown from callback " << e.what());
-            }
+            op.complete(ResultChecksumError, {});
+        } catch (const std::exception& e) {
+            LOG_ERROR(getName() << "Exception thrown from callback " << e.what());
         }
         releaseSemaphoreForSendOp(op);
         return true;
@@ -795,12 +858,10 @@ bool ProducerImpl::ackReceived(uint64_t sequenceId, MessageId& rawMessageId) {
         pendingMessagesQueue_.pop_front();
 
         lock.unlock();
-        if (op.sendCallback_) {
-            try {
-                op.sendCallback_(ResultOk, messageId);
-            } catch (const std::exception& e) {
-                LOG_ERROR(getName() << "Exception thrown from callback " << e.what());
-            }
+        try {
+            op.complete(ResultOk, messageId);
+        } catch (const std::exception& e) {
+            LOG_ERROR(getName() << "Exception thrown from callback " << e.what());
         }
         return true;
     }
diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h
index d29efed1a13ae..3dccf911f93b9 100644
--- a/pulsar-client-cpp/lib/ProducerImpl.h
+++ b/pulsar-client-cpp/lib/ProducerImpl.h
@@ -47,12 +47,13 @@ class PulsarFriend;
 
 class Producer;
 class MemoryLimitController;
+class TopicName;
 
 class ProducerImpl : public HandlerBase,
                      public std::enable_shared_from_this<ProducerImpl>,
                      public ProducerImplBase {
    public:
-    ProducerImpl(ClientImplPtr client, const std::string& topic,
+    ProducerImpl(ClientImplPtr client, const TopicName& topic,
                  const ProducerConfiguration& producerConfiguration, int32_t partition = -1);
     ~ProducerImpl();
 
@@ -83,6 +84,8 @@ class ProducerImpl : public HandlerBase,
 
     int32_t partition() const noexcept { return partition_; }
 
+    static int getNumOfChunks(uint32_t size, uint32_t maxMessageSize);
+
    protected:
     ProducerStatsBasePtr producerStatsBasePtr_;
 
@@ -115,8 +118,6 @@ class ProducerImpl : public HandlerBase,
     void handleCreateProducer(const ClientConnectionPtr& cnx, Result result,
                               const ResponseData& responseData);
 
-    void statsCallBackHandler(Result, const MessageId&, SendCallback, boost::posix_time::ptime);
-
     void handleClose(Result result, ResultCallback callback, ProducerImplPtr producer);
 
     void resendMessages(ClientConnectionPtr cnx);
@@ -125,12 +126,26 @@ class ProducerImpl : public HandlerBase,
     bool encryptMessage(proto::MessageMetadata& metadata, SharedBuffer& payload,
                         SharedBuffer& encryptedPayload);
 
+    void sendAsyncWithStatsUpdate(const Message& msg, const SendCallback& callback);
+
+    /**
+     * Reserve a spot in the messages queue before acquiring the ProducerImpl mutex. When the queue is full,
+     * this call will block until a spot is available if blockIfQueueIsFull is true. Otherwise, it will return
+     * ResultProducerQueueIsFull immediately.
+     *
+     * It also checks whether the memory could reach the limit after `payloadSize` is added. If so, this call
+     * will block until enough memory could be retained.
+     */
     Result canEnqueueRequest(uint32_t payloadSize);
+
     void releaseSemaphore(uint32_t payloadSize);
     void releaseSemaphoreForSendOp(const OpSendMsg& op);
 
     void cancelTimers();
 
+    bool isValidProducerState(const SendCallback& callback) const;
+    bool canAddToBatch(const Message& msg) const;
+
     typedef std::unique_lock<std::mutex> Lock;
 
     ProducerConfiguration conf_;
@@ -138,7 +153,7 @@ class ProducerImpl : public HandlerBase,
     std::unique_ptr<Semaphore> semaphore_;
     MessageQueue pendingMessagesQueue_;
 
-    int32_t partition_;  // -1 if topic is non-partitioned
+    const int32_t partition_;  // -1 if topic is non-partitioned
     std::string producerName_;
     bool userProvidedProducerName_;
     std::string producerStr_;
@@ -169,6 +184,7 @@ class ProducerImpl : public HandlerBase,
     uint32_t dataKeyGenIntervalSec_;
 
     MemoryLimitController& memoryLimitController_;
+    const bool chunkingEnabled_;
 };
 
 struct ProducerImplCmp {
diff --git a/pulsar-client-cpp/lib/SharedBuffer.h b/pulsar-client-cpp/lib/SharedBuffer.h
index d54429984b659..be889a7ee97ad 100644
--- a/pulsar-client-cpp/lib/SharedBuffer.h
+++ b/pulsar-client-cpp/lib/SharedBuffer.h
@@ -94,13 +94,13 @@ class SharedBuffer {
     /**
      * Return a shared buffer that include a portion of current buffer. No memory is copied
      */
-    SharedBuffer slice(uint32_t offset) {
+    SharedBuffer slice(uint32_t offset) const {
         SharedBuffer buf(*this);
         buf.consume(offset);
         return buf;
     }
 
-    SharedBuffer slice(uint32_t offset, uint32_t length) {
+    SharedBuffer slice(uint32_t offset, uint32_t length) const {
         SharedBuffer buf(*this);
         buf.consume(offset);
         assert(buf.readableBytes() >= length);
@@ -165,7 +165,7 @@ class SharedBuffer {
     }
 
     // Return current writer index
-    uint32_t writerIndex() { return writeIdx_; }
+    uint32_t writerIndex() const noexcept { return writeIdx_; }
 
     // skip writerIndex
     void skipBytes(uint32_t size) {
@@ -180,7 +180,7 @@ class SharedBuffer {
     }
 
     // Return current reader index
-    uint32_t readerIndex() { return readIdx_; }
+    uint32_t readerIndex() const noexcept { return readIdx_; }
 
     // set readerIndex
     void setReaderIndex(uint32_t index) {
diff --git a/pulsar-client-cpp/lib/TopicName.cc b/pulsar-client-cpp/lib/TopicName.cc
index a56b9795f396e..2e6232cf001de 100644
--- a/pulsar-client-cpp/lib/TopicName.cc
+++ b/pulsar-client-cpp/lib/TopicName.cc
@@ -216,7 +216,7 @@ std::string TopicName::getLookupName() {
     return ss.str();
 }
 
-std::string TopicName::toString() {
+std::string TopicName::toString() const {
     std::stringstream ss;
     std::string seperator("/");
     if (isV2Topic_ && cluster_.empty()) {
@@ -230,7 +230,7 @@ std::string TopicName::toString() {
 
 bool TopicName::isPersistent() const { return this->domain_ == TopicDomain::Persistent; }
 
-const std::string TopicName::getTopicPartitionName(unsigned int partition) {
+std::string TopicName::getTopicPartitionName(unsigned int partition) const {
     std::stringstream topicPartitionName;
     // make this topic name as well
     topicPartitionName << toString() << PartitionedProducerImpl::PARTITION_NAME_SUFFIX << partition;
diff --git a/pulsar-client-cpp/lib/TopicName.h b/pulsar-client-cpp/lib/TopicName.h
index 248001c172a41..1d5deab553dca 100644
--- a/pulsar-client-cpp/lib/TopicName.h
+++ b/pulsar-client-cpp/lib/TopicName.h
@@ -55,14 +55,14 @@ class PULSAR_PUBLIC TopicName : public ServiceUnitId {
     std::string getNamespacePortion();
     std::string getLocalName();
     std::string getEncodedLocalName();
-    std::string toString();
+    std::string toString() const;
     bool isPersistent() const;
     NamespaceNamePtr getNamespaceName();
     int getPartitionIndex() const noexcept { return partition_; }
     static std::shared_ptr<TopicName> get(const std::string& topicName);
     bool operator==(const TopicName& other);
     static std::string getEncodedName(const std::string& nameBeforeEncoding);
-    const std::string getTopicPartitionName(unsigned int partition);
+    std::string getTopicPartitionName(unsigned int partition) const;
     static int getPartitionIndex(const std::string& topic);
 
    private:
diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsBase.h b/pulsar-client-cpp/lib/stats/ProducerStatsBase.h
index 494303b5dfe82..0ae16d1769c7c 100644
--- a/pulsar-client-cpp/lib/stats/ProducerStatsBase.h
+++ b/pulsar-client-cpp/lib/stats/ProducerStatsBase.h
@@ -27,7 +27,7 @@ namespace pulsar {
 class ProducerStatsBase {
    public:
     virtual void messageSent(const Message& msg) = 0;
-    virtual void messageReceived(Result&, boost::posix_time::ptime&) = 0;
+    virtual void messageReceived(Result, const boost::posix_time::ptime&) = 0;
     virtual ~ProducerStatsBase(){};
 };
 
diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h b/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h
index f81c8aa1a7944..6568c07487719 100644
--- a/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h
+++ b/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h
@@ -25,7 +25,7 @@ namespace pulsar {
 class ProducerStatsDisabled : public ProducerStatsBase {
    public:
     virtual void messageSent(const Message& msg){};
-    virtual void messageReceived(Result&, boost::posix_time::ptime&){};
+    virtual void messageReceived(Result, const boost::posix_time::ptime&){};
 };
 }  // namespace pulsar
 #endif  // PULSAR_PRODUCER_STATS_DISABLED_HEADER
diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc
index af7ae4b9c0440..811c0e16753fe 100644
--- a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc
+++ b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc
@@ -93,7 +93,7 @@ void ProducerStatsImpl::messageSent(const Message& msg) {
     totalBytesSent_ += msg.getLength();
 }
 
-void ProducerStatsImpl::messageReceived(Result& res, boost::posix_time::ptime& publishTime) {
+void ProducerStatsImpl::messageReceived(Result res, const boost::posix_time::ptime& publishTime) {
     boost::posix_time::ptime currentTime = boost::posix_time::microsec_clock::universal_time();
     double diffInMicros = (currentTime - publishTime).total_microseconds();
     Lock lock(mutex_);
diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h
index e82628b140abe..27ffacc81a519 100644
--- a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h
+++ b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h
@@ -82,7 +82,7 @@ class ProducerStatsImpl : public std::enable_shared_from_this<ProducerStatsImpl>
 
     void messageSent(const Message&);
 
-    void messageReceived(Result&, boost::posix_time::ptime&);
+    void messageReceived(Result, const boost::posix_time::ptime&);
 
     ~ProducerStatsImpl();
 
diff --git a/pulsar-client-cpp/test-conf/standalone-ssl.conf b/pulsar-client-cpp/test-conf/standalone-ssl.conf
index 90c48228f9271..5c94e58720deb 100644
--- a/pulsar-client-cpp/test-conf/standalone-ssl.conf
+++ b/pulsar-client-cpp/test-conf/standalone-ssl.conf
@@ -300,4 +300,7 @@ defaultNumPartitions=1
 globalZookeeperServers={{ zookeeper_servers }}
 
 # Deprecated. Use brokerDeleteInactiveTopicsFrequencySeconds
-brokerServicePurgeInactiveFrequencyInSeconds=60
\ No newline at end of file
+brokerServicePurgeInactiveFrequencyInSeconds=60
+
+# Given a specific limit of the max message size
+maxMessageSize=1024000
diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
index be6577523ed7d..da5c60952dd18 100644
--- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
+++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
@@ -602,7 +602,7 @@ TEST(BasicEndToEndTest, testMessageTooBig) {
     Result result = client.createProducer(topicName, conf, producer);
     ASSERT_EQ(ResultOk, result);
 
-    int size = Commands::DefaultMaxMessageSize + 1000 * 100;
+    int size = ClientConnection::getMaxMessageSize() + 1000 * 100;
     char *content = new char[size];
     memset(content, 0, size);
     Message msg = MessageBuilder().setAllocatedContent(content, size).build();
@@ -610,7 +610,7 @@ TEST(BasicEndToEndTest, testMessageTooBig) {
     ASSERT_EQ(ResultMessageTooBig, result);
 
     // Anything up to MaxMessageSize should be allowed
-    size = Commands::DefaultMaxMessageSize;
+    size = ClientConnection::getMaxMessageSize();
     msg = MessageBuilder().setAllocatedContent(content, size).build();
     result = producer.send(msg);
     ASSERT_EQ(ResultOk, result);
@@ -1156,7 +1156,7 @@ TEST(BasicEndToEndTest, testProduceMessageSize) {
     result = producerFuture.get(producer2);
     ASSERT_EQ(ResultOk, result);
 
-    int size = Commands::DefaultMaxMessageSize + 1000 * 100;
+    int size = ClientConnection::getMaxMessageSize() + 1000 * 100;
     char *content = new char[size];
     memset(content, 0, size);
     Message msg = MessageBuilder().setAllocatedContent(content, size).build();
@@ -1208,7 +1208,7 @@ TEST(BasicEndToEndTest, testBigMessageSizeBatching) {
     result = client.createProducer(topicName, conf2, producer2);
     ASSERT_EQ(ResultOk, result);
 
-    int size = Commands::DefaultMaxMessageSize + 1000 * 100;
+    int size = ClientConnection::getMaxMessageSize() + 1000 * 100;
     char *content = new char[size];
     memset(content, 0, size);
     Message msg = MessageBuilder().setAllocatedContent(content, size).build();
diff --git a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
index 199b50c0f7174..57ed0ec7c79c6 100644
--- a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
+++ b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc
@@ -59,6 +59,8 @@ TEST(ConsumerConfigurationTest, testDefaultConfig) {
     ASSERT_EQ(conf.isReplicateSubscriptionStateEnabled(), false);
     ASSERT_EQ(conf.getProperties().empty(), true);
     ASSERT_EQ(conf.getPriorityLevel(), 0);
+    ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 100);
+    ASSERT_EQ(conf.isAutoOldestChunkedMessageOnQueueFull(), false);
 }
 
 TEST(ConsumerConfigurationTest, testCustomConfig) {
@@ -139,6 +141,12 @@ TEST(ConsumerConfigurationTest, testCustomConfig) {
 
     conf.setPriorityLevel(1);
     ASSERT_EQ(conf.getPriorityLevel(), 1);
+
+    conf.setMaxPendingChunkedMessage(500);
+    ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 500);
+
+    conf.setAutoOldestChunkedMessageOnQueueFull(true);
+    ASSERT_TRUE(conf.isAutoOldestChunkedMessageOnQueueFull());
 }
 
 TEST(ConsumerConfigurationTest, testReadCompactPersistentExclusive) {
diff --git a/pulsar-client-cpp/tests/MapCacheTest.cc b/pulsar-client-cpp/tests/MapCacheTest.cc
new file mode 100644
index 0000000000000..12a89ee17be34
--- /dev/null
+++ b/pulsar-client-cpp/tests/MapCacheTest.cc
@@ -0,0 +1,78 @@
+/**
+ * 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 <gtest/gtest.h>
+#include <lib/MapCache.h>
+
+using namespace pulsar;
+
+struct MoveOnlyInt {
+    int x = 0;
+
+    MoveOnlyInt() = default;
+    MoveOnlyInt(int xx) : x(xx) {}
+    MoveOnlyInt(const MoveOnlyInt&) = delete;
+    MoveOnlyInt(MoveOnlyInt&& rhs) noexcept : x(rhs.x) {}
+
+    bool operator=(const MoveOnlyInt& rhs) const { return x == rhs.x; }
+};
+
+TEST(MapCacheTest, testPutIfAbsent) {
+    MapCache<int, MoveOnlyInt> cache;
+
+    ASSERT_NE(cache.putIfAbsent(1, {100}), cache.end());
+    ASSERT_EQ(cache.putIfAbsent(1, {200}), cache.end());
+    auto it = cache.find(1);
+    ASSERT_NE(it, cache.end());
+    ASSERT_EQ(it->second.x, 100);
+
+    cache.remove(1);
+    ASSERT_EQ(cache.find(1), cache.end());
+}
+
+TEST(MapCacheTest, testRemoveOldestValues) {
+    MapCache<int, MoveOnlyInt> cache;
+    cache.putIfAbsent(1, {200});
+    cache.putIfAbsent(2, {210});
+    cache.putIfAbsent(3, {220});
+    ASSERT_EQ(cache.getKeys(), (std::vector<int>{1, 2, 3}));
+
+    std::vector<int> removedValues;
+    cache.removeOldestValues(2, [&removedValues](const int& key, const MoveOnlyInt& value) {
+        removedValues.emplace_back(value.x);
+    });
+    ASSERT_EQ(removedValues, (std::vector<int>{200, 210}));
+
+    ASSERT_EQ(cache.getKeys(), (std::vector<int>{3}));
+    ASSERT_EQ(cache.size(), 1);
+    auto it = cache.find(3);
+    ASSERT_NE(it, cache.end());
+    ASSERT_EQ(it->second.x, 220);
+}
+
+TEST(MapCacheTest, testRemoveAllValues) {
+    MapCache<int, MoveOnlyInt> cache;
+    cache.putIfAbsent(1, {300});
+    cache.putIfAbsent(2, {310});
+    cache.putIfAbsent(3, {320});
+
+    // removeOldestValues works well even if the argument is greater than the size of keys
+    cache.removeOldestValues(10000, nullptr);
+    ASSERT_TRUE(cache.getKeys().empty());
+    ASSERT_EQ(cache.size(), 0);
+}
diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc
new file mode 100644
index 0000000000000..ae0114cefb162
--- /dev/null
+++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc
@@ -0,0 +1,137 @@
+/**
+ * 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 <ctime>
+#include <random>
+
+#include <pulsar/Client.h>
+#include <gtest/gtest.h>
+#include "lib/LogUtils.h"
+#include "PulsarFriend.h"
+
+DECLARE_LOG_OBJECT()
+
+using namespace pulsar;
+
+static const std::string lookupUrl = "pulsar://localhost:6650";
+
+// See the `maxMessageSize` config in test-conf/standalone-ssl.conf
+static constexpr size_t maxMessageSize = 1024000;
+
+static std::string toString(CompressionType compressionType) {
+    switch (compressionType) {
+        case CompressionType::CompressionNone:
+            return "None";
+        case CompressionType::CompressionLZ4:
+            return "LZ4";
+        case CompressionType::CompressionZLib:
+            return "ZLib";
+        case CompressionType::CompressionZSTD:
+            return "ZSTD";
+        case CompressionType::CompressionSNAPPY:
+            return "SNAPPY";
+        default:
+            return "Unknown (" + std::to_string(compressionType) + ")";
+    }
+}
+
+inline std::string createLargeMessage() {
+    std::string largeMessage(maxMessageSize * 3, 'a');
+    std::default_random_engine e(time(nullptr));
+    std::uniform_int_distribution<unsigned> u(0, 25);
+    for (size_t i = 0; i < largeMessage.size(); i++) {
+        largeMessage[i] = 'a' + u(e);
+    }
+    return largeMessage;
+}
+
+class MessageChunkingTest : public ::testing::TestWithParam<CompressionType> {
+   public:
+    static std::string largeMessage;
+
+    void TearDown() override { client_.close(); }
+
+    void createProducer(const std::string& topic, Producer& producer) {
+        ProducerConfiguration conf;
+        conf.setBatchingEnabled(false);
+        conf.setChunkingEnabled(true);
+        conf.setCompressionType(GetParam());
+        LOG_INFO("Create producer to topic: " << topic
+                                              << ", compression: " << toString(conf.getCompressionType()));
+        ASSERT_EQ(ResultOk, client_.createProducer(topic, conf, producer));
+    }
+
+    void createConsumer(const std::string& topic, Consumer& consumer) {
+        ASSERT_EQ(ResultOk, client_.subscribe(topic, "my-sub", consumer));
+    }
+
+   private:
+    Client client_{lookupUrl};
+};
+
+std::string MessageChunkingTest::largeMessage = createLargeMessage();
+
+TEST_F(MessageChunkingTest, testInvalidConfig) {
+    Client client(lookupUrl);
+    ProducerConfiguration conf;
+    conf.setBatchingEnabled(true);
+    conf.setChunkingEnabled(true);
+    Producer producer;
+    ASSERT_THROW(client.createProducer("xxx", conf, producer), std::invalid_argument);
+    client.close();
+}
+
+TEST_P(MessageChunkingTest, testEndToEnd) {
+    const std::string topic =
+        "MessageChunkingTest-EndToEnd-" + toString(GetParam()) + std::to_string(time(nullptr));
+    Consumer consumer;
+    createConsumer(topic, consumer);
+    Producer producer;
+    createProducer(topic, producer);
+
+    constexpr int numMessages = 10;
+
+    std::vector<MessageId> sendMessageIds;
+    for (int i = 0; i < numMessages; i++) {
+        MessageId messageId;
+        ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), messageId));
+        LOG_INFO("Send " << i << " to " << messageId);
+        sendMessageIds.emplace_back(messageId);
+    }
+
+    Message msg;
+    std::vector<MessageId> receivedMessageIds;
+    for (int i = 0; i < numMessages; i++) {
+        ASSERT_EQ(ResultOk, consumer.receive(msg, 3000));
+        LOG_INFO("Receive " << msg.getLength() << " bytes from " << msg.getMessageId());
+        ASSERT_EQ(msg.getDataAsString(), largeMessage);
+        receivedMessageIds.emplace_back(msg.getMessageId());
+    }
+    ASSERT_EQ(receivedMessageIds, sendMessageIds);
+    ASSERT_EQ(receivedMessageIds.front().ledgerId(), receivedMessageIds.front().ledgerId());
+    ASSERT_GT(receivedMessageIds.back().entryId(), numMessages);
+
+    // Verify the cache has been cleared
+    auto& chunkedMessageCache = PulsarFriend::getChunkedMessageCache(consumer);
+    ASSERT_EQ(chunkedMessageCache.size(), 0);
+}
+
+// The CI env is Ubuntu 16.04, the gtest-dev version is 1.8.0 that doesn't have INSTANTIATE_TEST_SUITE_P
+INSTANTIATE_TEST_CASE_P(Pulsar, MessageChunkingTest,
+                        ::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD,
+                                          CompressionSNAPPY));
diff --git a/pulsar-client-cpp/tests/ProducerConfigurationTest.cc b/pulsar-client-cpp/tests/ProducerConfigurationTest.cc
index b88f6e41890d8..5c541295ff966 100644
--- a/pulsar-client-cpp/tests/ProducerConfigurationTest.cc
+++ b/pulsar-client-cpp/tests/ProducerConfigurationTest.cc
@@ -46,6 +46,7 @@ TEST(ProducerConfigurationTest, testDefaultConfig) {
     ASSERT_EQ(conf.isEncryptionEnabled(), false);
     ASSERT_EQ(conf.getEncryptionKeys(), std::set<std::string>{});
     ASSERT_EQ(conf.getProperties().empty(), true);
+    ASSERT_EQ(conf.isChunkingEnabled(), false);
 }
 
 class MockMessageRoutingPolicy : public MessageRoutingPolicy {
@@ -129,4 +130,7 @@ TEST(ProducerConfigurationTest, testCustomConfig) {
     conf.setProperty("k1", "v1");
     ASSERT_EQ(conf.getProperties()["k1"], "v1");
     ASSERT_EQ(conf.hasProperty("k1"), true);
+
+    conf.setChunkingEnabled(true);
+    ASSERT_EQ(conf.isChunkingEnabled(), true);
 }
diff --git a/pulsar-client-cpp/tests/ProducerTest.cc b/pulsar-client-cpp/tests/ProducerTest.cc
index 210f01345d4af..3ee8def37dbaf 100644
--- a/pulsar-client-cpp/tests/ProducerTest.cc
+++ b/pulsar-client-cpp/tests/ProducerTest.cc
@@ -26,6 +26,7 @@
 #include "lib/Utils.h"
 #include "lib/Latch.h"
 #include "lib/LogUtils.h"
+#include "lib/ProducerImpl.h"
 DECLARE_LOG_OBJECT()
 
 using namespace pulsar;
@@ -240,4 +241,12 @@ TEST(ProducerTest, testSendAsyncCloseAsyncConcurrentlyWithLazyProducers) {
         client.close();
         LOG_INFO("End of run " << run);
     }
-}
\ No newline at end of file
+}
+
+TEST(ProducerTest, testGetNumOfChunks) {
+    ASSERT_EQ(ProducerImpl::getNumOfChunks(11, 5), 3);
+    ASSERT_EQ(ProducerImpl::getNumOfChunks(10, 5), 2);
+    ASSERT_EQ(ProducerImpl::getNumOfChunks(8, 5), 2);
+    ASSERT_EQ(ProducerImpl::getNumOfChunks(4, 5), 1);
+    ASSERT_EQ(ProducerImpl::getNumOfChunks(1, 0), 1);
+}
diff --git a/pulsar-client-cpp/tests/PulsarFriend.h b/pulsar-client-cpp/tests/PulsarFriend.h
index aed7096366ad8..c7aa3a67bb957 100644
--- a/pulsar-client-cpp/tests/PulsarFriend.h
+++ b/pulsar-client-cpp/tests/PulsarFriend.h
@@ -79,6 +79,12 @@ class PulsarFriend {
         return std::static_pointer_cast<ConsumerImpl>(consumer.impl_);
     }
 
+    static decltype(ConsumerImpl::chunkedMessageCache_) & getChunkedMessageCache(Consumer consumer) {
+        auto consumerImpl = getConsumerImplPtr(consumer);
+        ConsumerImpl::Lock lock(consumerImpl->chunkProcessMutex_);
+        return consumerImpl->chunkedMessageCache_;
+    }
+
     static std::shared_ptr<PartitionedConsumerImpl> getPartitionedConsumerImplPtr(Consumer consumer) {
         return std::static_pointer_cast<PartitionedConsumerImpl>(consumer.impl_);
     }