Skip to content

[fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages #604

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
geniusjoe wants to merge 1 commit into
base: main
Choose a base branch
from
+190 −4
Open

[fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages #604

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 24 additions & 2 deletions lib/ConsumerImpl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1490,7 +1490,16 @@ std::pair<MessageId, bool> ConsumerImpl::prepareCumulativeAck(const MessageId& m

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); }
Expand Down Expand Up @@ -1651,7 +1660,19 @@ void ConsumerImpl::redeliverMessages(const std::set<MessageId>& messageIds) {
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 {
Expand Down Expand Up @@ -2037,6 +2058,7 @@ void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, const Proces
producerConfiguration.setSchema(config_.getSchema());
producerConfiguration.setBlockIfQueueFull(false);
producerConfiguration.setBatchingEnabled(false);
producerConfiguration.setChunkingEnabled(true);
producerConfiguration.impl_->initialSubscriptionName =
deadLetterPolicy_.getInitialSubscriptionName();
ClientImplPtr client = client_.lock();
Expand Down
18 changes: 16 additions & 2 deletions lib/UnAckedMessageTrackerEnabled.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

#include <functional>

#include "ChunkMessageIdImpl.h"
#include "ClientConnection.h"
#include "ClientImpl.h"
#include "ConsumerImplBase.h"
Expand Down Expand Up @@ -108,7 +109,18 @@ void UnAckedMessageTrackerEnabled::start() { timeoutHandler(); }

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;
Expand All @@ -125,7 +137,9 @@ bool UnAckedMessageTrackerEnabled::isEmpty() {

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);
Expand Down
150 changes: 150 additions & 0 deletions tests/MessageChunkingTest.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -454,6 +457,153 @@ TEST_P(MessageChunkingTest, testResendChunkWithAckHoleMessages) {
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,
Expand Down