[fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages (#604)
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index 3337e81..657e1e6 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -1490,7 +1490,16 @@
void ConsumerImpl::negativeAcknowledge(const MessageId& messageId) {
unAckedMessageTrackerPtr_->remove(messageId);
- negativeAcksTracker_->add(messageId);
+ // If it's a ChunkMessageId, expand all chunk entries and nack them individually,
+ // so that the broker can redeliver all chunks
+ if (auto chunkMessageId =
+ std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(messageId))) {
+ for (const auto& chunkId : chunkMessageId->getChunkedMessageIds()) {
+ negativeAcksTracker_->add(chunkId);
+ }
+ } else {
+ negativeAcksTracker_->add(messageId);
+ }
}
void ConsumerImpl::disconnectConsumer() { disconnectConsumer(std::nullopt); }
@@ -1651,7 +1660,19 @@
ClientConnectionPtr cnx = getCnx().lock();
if (cnx) {
if (cnx->getServerProtocolVersion() >= proto::v2) {
- cnx->sendCommand(Commands::newRedeliverUnacknowledgedMessages(consumerId_, messageIds));
+ // Expand ChunkMessageIds into all chunk entries to ensure the broker
+ // can redeliver all chunks
+ std::set<MessageId> expandedMsgIds;
+ for (const auto& msgId : messageIds) {
+ if (auto chunkMsgId =
+ std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(msgId))) {
+ const auto& chunkIds = chunkMsgId->getChunkedMessageIds();
+ expandedMsgIds.insert(chunkIds.begin(), chunkIds.end());
+ } else {
+ expandedMsgIds.insert(msgId);
+ }
+ }
+ cnx->sendCommand(Commands::newRedeliverUnacknowledgedMessages(consumerId_, expandedMsgIds));
LOG_DEBUG("Sending RedeliverUnacknowledgedMessages command for Consumer - " << getConsumerId());
}
} else {
@@ -2037,6 +2058,7 @@
producerConfiguration.setSchema(config_.getSchema());
producerConfiguration.setBlockIfQueueFull(false);
producerConfiguration.setBatchingEnabled(false);
+ producerConfiguration.setChunkingEnabled(true);
producerConfiguration.impl_->initialSubscriptionName =
deadLetterPolicy_.getInitialSubscriptionName();
ClientImplPtr client = client_.lock();
diff --git a/lib/UnAckedMessageTrackerEnabled.cc b/lib/UnAckedMessageTrackerEnabled.cc
index 1c7b20e..b366460 100644
--- a/lib/UnAckedMessageTrackerEnabled.cc
+++ b/lib/UnAckedMessageTrackerEnabled.cc
@@ -20,6 +20,7 @@
#include <functional>
+#include "ChunkMessageIdImpl.h"
#include "ClientConnection.h"
#include "ClientImpl.h"
#include "ConsumerImplBase.h"
@@ -108,7 +109,18 @@
bool UnAckedMessageTrackerEnabled::add(const MessageId& msgId) {
std::lock_guard<std::recursive_mutex> acquire(lock_);
- auto id = discardBatch(msgId);
+ // For ChunkMessageId, skip discardBatch to preserve the original ChunkMessageIdImpl.
+ //
+ // ChunkMessageIdImpl stores all chunk entries internally (chunkedMessageIds_), and its
+ // ledgerId/entryId are set to the last chunk's position. If discardBatch is applied, a
+ // new plain MessageIdImpl would be created, losing all chunk entries information.
+ //
+ // Although the set/map key only reflects the last chunk's position, the MessageId object
+ // itself retains the full ChunkMessageIdImpl via its impl_ pointer. So when ackTimeout
+ // triggers and the MessageId is passed to redeliverMessages(), it can be expanded into
+ // all chunk entries for redelivery, ensuring the broker redelivers the complete chunked
+ // message.
+ auto id = std::dynamic_pointer_cast<ChunkMessageIdImpl>(msgId.impl_) ? msgId : discardBatch(msgId);
if (messageIdPartitionMap.count(id) == 0) {
std::set<MessageId>& partition = timePartitions.back();
bool emplace = messageIdPartitionMap.emplace(id, partition).second;
@@ -125,7 +137,9 @@
bool UnAckedMessageTrackerEnabled::remove(const MessageId& msgId) {
std::lock_guard<std::recursive_mutex> acquire(lock_);
- auto id = discardBatch(msgId);
+ // Keep consistent with add(): skip discardBatch for ChunkMessageId to ensure
+ // the same key is used for lookup and removal.
+ auto id = std::dynamic_pointer_cast<ChunkMessageIdImpl>(msgId.impl_) ? msgId : discardBatch(msgId);
bool removed = false;
std::map<MessageId, std::set<MessageId>&>::iterator exist = messageIdPartitionMap.find(id);
diff --git a/tests/MessageChunkingTest.cc b/tests/MessageChunkingTest.cc
index 2421d40..aa684ac 100644
--- a/tests/MessageChunkingTest.cc
+++ b/tests/MessageChunkingTest.cc
@@ -18,14 +18,17 @@
*/
#include <gtest/gtest.h>
#include <pulsar/Client.h>
+#include <pulsar/DeadLetterPolicyBuilder.h>
#include <pulsar/MessageIdBuilder.h>
#include <ctime>
#include <random>
+#include <sstream>
#include "PulsarFriend.h"
#include "WaitUtils.h"
#include "lib/ChunkMessageIdImpl.h"
+#include "lib/ConsumerImpl.h"
#include "lib/LogUtils.h"
DECLARE_LOG_OBJECT()
@@ -454,6 +457,153 @@
consumer.close();
}
+// Aligned with Go TestChunkAckAndNAck and Java testNegativeAckChunkedMessage
+TEST_P(MessageChunkingTest, testNegativeAckChunkedMessage) {
+ if (toString(GetParam()) != "None") {
+ return;
+ }
+ const std::string topic =
+ "MessageChunkingTest-testNegativeAckChunkedMessage-" + std::to_string(time(nullptr));
+
+ Consumer consumer;
+ ConsumerConfiguration consumerConf;
+ consumerConf.setConsumerType(ConsumerShared);
+ consumerConf.setNegativeAckRedeliveryDelayMs(1000);
+ createConsumer(topic, consumer, consumerConf);
+
+ Producer producer;
+ createProducer(topic, producer);
+
+ // Send a chunked message
+ MessageId sendMsgId;
+ ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMsgId));
+
+ // Receive and nack
+ Message msg;
+ ASSERT_EQ(ResultOk, consumer.receive(msg, 5000));
+ ASSERT_EQ(msg.getDataAsString(), largeMessage);
+ consumer.negativeAcknowledge(msg);
+
+ // The message should be redelivered after nack delay
+ Message redeliveredMsg;
+ ASSERT_EQ(ResultOk, consumer.receive(redeliveredMsg, 5000));
+ ASSERT_EQ(redeliveredMsg.getDataAsString(), largeMessage);
+ consumer.acknowledge(redeliveredMsg);
+
+ // Verify no more messages
+ Message noMsg;
+ ASSERT_NE(ResultOk, consumer.receive(noMsg, 2000));
+
+ producer.close();
+ consumer.close();
+}
+
+// Aligned with Java testLargeMessageAckTimeOut
+TEST_P(MessageChunkingTest, testAckTimeoutChunkedMessage) {
+ if (toString(GetParam()) != "None") {
+ return;
+ }
+ const std::string topic =
+ "MessageChunkingTest-testAckTimeoutChunkedMessage-" + std::to_string(time(nullptr));
+
+ Consumer consumer;
+ ConsumerConfiguration consumerConf;
+ consumerConf.setConsumerType(ConsumerShared);
+ // Set ack timeout to 2 seconds
+ PulsarFriend::setConsumerUnAckMessagesTimeoutMs(consumerConf, 2000);
+ createConsumer(topic, consumer, consumerConf);
+
+ Producer producer;
+ createProducer(topic, producer);
+
+ // Send a chunked message
+ MessageId sendMsgId;
+ ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMsgId));
+
+ // Receive but do NOT acknowledge - let ack timeout trigger redelivery
+ Message msg;
+ ASSERT_EQ(ResultOk, consumer.receive(msg, 5000));
+ ASSERT_EQ(msg.getDataAsString(), largeMessage);
+
+ // Wait for ack timeout to trigger redelivery
+ // The message should be redelivered after ack timeout (2s)
+ Message redeliveredMsg;
+ ASSERT_EQ(ResultOk, consumer.receive(redeliveredMsg, 5000));
+ ASSERT_EQ(redeliveredMsg.getDataAsString(), largeMessage);
+ consumer.acknowledge(redeliveredMsg);
+
+ // Verify no more messages
+ Message noMsg;
+ ASSERT_NE(ResultOk, consumer.receive(noMsg, 2000));
+
+ producer.close();
+ consumer.close();
+}
+
+TEST_P(MessageChunkingTest, testChunkedMessageDLQ) {
+ if (toString(GetParam()) != "None") {
+ return;
+ }
+ const std::string topic = "persistent://public/default/MessageChunkingTest-testChunkedMessageDLQ-" +
+ std::to_string(time(nullptr));
+ const std::string subName = "my-sub";
+ const std::string dlqTopic = topic + "-" + subName + "-DLQ";
+
+ Client client(lookupUrl);
+
+ auto dlqPolicy =
+ DeadLetterPolicyBuilder().maxRedeliverCount(2).initialSubscriptionName("dlq-init-sub").build();
+
+ Consumer consumer;
+ ConsumerConfiguration consumerConf;
+ consumerConf.setConsumerType(ConsumerShared);
+ consumerConf.setNegativeAckRedeliveryDelayMs(100);
+ consumerConf.setDeadLetterPolicy(dlqPolicy);
+ ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConf, consumer));
+
+ // Subscribe to DLQ topic to verify messages arrive there
+ Consumer dlqConsumer;
+ ConsumerConfiguration dlqConsumerConf;
+ dlqConsumerConf.setConsumerType(ConsumerShared);
+ ASSERT_EQ(ResultOk, client.subscribe(dlqTopic, "dlq-sub", dlqConsumerConf, dlqConsumer));
+
+ Producer producer;
+ createProducer(topic, producer);
+
+ // Send a chunked message
+ MessageId sendMsgId;
+ ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMsgId));
+
+ // Nack the message maxRedeliverCount + 1 times to trigger DLQ
+ Message msg;
+ for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() + 1; i++) {
+ ASSERT_EQ(ResultOk, consumer.receive(msg, 5000));
+ ASSERT_EQ(msg.getDataAsString(), largeMessage);
+ consumer.negativeAcknowledge(msg);
+ }
+
+ // Verify the message arrives in DLQ with correct content
+ Message dlqMsg;
+ ASSERT_EQ(ResultOk, dlqConsumer.receive(dlqMsg, 10000));
+ ASSERT_EQ(dlqMsg.getDataAsString(), largeMessage);
+ std::stringstream expectedOriginMsgId;
+ expectedOriginMsgId << sendMsgId;
+ ASSERT_EQ(dlqMsg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID), expectedOriginMsgId.str());
+ ASSERT_EQ(dlqMsg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC), topic);
+
+ // Verify no more messages in DLQ
+ Message noMsg;
+ ASSERT_NE(ResultOk, dlqConsumer.receive(noMsg, 2000));
+
+ // Verify original consumer has no more messages (message was acked after DLQ send)
+ ASSERT_NE(ResultOk, consumer.receive(noMsg, 2000));
+
+ producer.close();
+ consumer.close();
+ dlqConsumer.close();
+ client.close();
+}
+
// 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,
