Skip to content

Commit 1d08b2b

Browse files
geniusjoegeniusjoe
authored
[fix][client] Support nack, ackTimeout redelivery and dlq for chunked messages (#604)
1 parent 7892883 commit 1d08b2b

3 files changed

Lines changed: 190 additions & 4 deletions

File tree

‎lib/ConsumerImpl.cc‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1490,7 +1490,16 @@ std::pair<MessageId, bool> ConsumerImpl::prepareCumulativeAck(const MessageId& m
14901490

14911491
void ConsumerImpl::negativeAcknowledge(const MessageId& messageId) {
14921492
unAckedMessageTrackerPtr_->remove(messageId);
1493-
negativeAcksTracker_->add(messageId);
1493+
// If it's a ChunkMessageId, expand all chunk entries and nack them individually,
1494+
// so that the broker can redeliver all chunks
1495+
if (auto chunkMessageId =
1496+
std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(messageId))) {
1497+
for (const auto& chunkId : chunkMessageId->getChunkedMessageIds()) {
1498+
negativeAcksTracker_->add(chunkId);
1499+
}
1500+
} else {
1501+
negativeAcksTracker_->add(messageId);
1502+
}
14941503
}
14951504

14961505
void ConsumerImpl::disconnectConsumer() { disconnectConsumer(std::nullopt); }
@@ -1651,7 +1660,19 @@ void ConsumerImpl::redeliverMessages(const std::set<MessageId>& messageIds) {
16511660
ClientConnectionPtr cnx = getCnx().lock();
16521661
if (cnx) {
16531662
if (cnx->getServerProtocolVersion() >= proto::v2) {
1654-
cnx->sendCommand(Commands::newRedeliverUnacknowledgedMessages(consumerId_, messageIds));
1663+
// Expand ChunkMessageIds into all chunk entries to ensure the broker
1664+
// can redeliver all chunks
1665+
std::set<MessageId> expandedMsgIds;
1666+
for (const auto& msgId : messageIds) {
1667+
if (auto chunkMsgId =
1668+
std::dynamic_pointer_cast<ChunkMessageIdImpl>(Commands::getMessageIdImpl(msgId))) {
1669+
const auto& chunkIds = chunkMsgId->getChunkedMessageIds();
1670+
expandedMsgIds.insert(chunkIds.begin(), chunkIds.end());
1671+
} else {
1672+
expandedMsgIds.insert(msgId);
1673+
}
1674+
}
1675+
cnx->sendCommand(Commands::newRedeliverUnacknowledgedMessages(consumerId_, expandedMsgIds));
16551676
LOG_DEBUG("Sending RedeliverUnacknowledgedMessages command for Consumer - " << getConsumerId());
16561677
}
16571678
} else {
@@ -2037,6 +2058,7 @@ void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, const Proces
20372058
producerConfiguration.setSchema(config_.getSchema());
20382059
producerConfiguration.setBlockIfQueueFull(false);
20392060
producerConfiguration.setBatchingEnabled(false);
2061+
producerConfiguration.setChunkingEnabled(true);
20402062
producerConfiguration.impl_->initialSubscriptionName =
20412063
deadLetterPolicy_.getInitialSubscriptionName();
20422064
ClientImplPtr client = client_.lock();

‎lib/UnAckedMessageTrackerEnabled.cc‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020

2121
#include <functional>
2222

23+
#include "ChunkMessageIdImpl.h"
2324
#include "ClientConnection.h"
2425
#include "ClientImpl.h"
2526
#include "ConsumerImplBase.h"
@@ -108,7 +109,18 @@ void UnAckedMessageTrackerEnabled::start() { timeoutHandler(); }
108109

109110
bool UnAckedMessageTrackerEnabled::add(const MessageId& msgId) {
110111
std::lock_guard<std::recursive_mutex> acquire(lock_);
111-
auto id = discardBatch(msgId);
112+
// For ChunkMessageId, skip discardBatch to preserve the original ChunkMessageIdImpl.
113+
//
114+
// ChunkMessageIdImpl stores all chunk entries internally (chunkedMessageIds_), and its
115+
// ledgerId/entryId are set to the last chunk's position. If discardBatch is applied, a
116+
// new plain MessageIdImpl would be created, losing all chunk entries information.
117+
//
118+
// Although the set/map key only reflects the last chunk's position, the MessageId object
119+
// itself retains the full ChunkMessageIdImpl via its impl_ pointer. So when ackTimeout
120+
// triggers and the MessageId is passed to redeliverMessages(), it can be expanded into
121+
// all chunk entries for redelivery, ensuring the broker redelivers the complete chunked
122+
// message.
123+
auto id = std::dynamic_pointer_cast<ChunkMessageIdImpl>(msgId.impl_) ? msgId : discardBatch(msgId);
112124
if (messageIdPartitionMap.count(id) == 0) {
113125
std::set<MessageId>& partition = timePartitions.back();
114126
bool emplace = messageIdPartitionMap.emplace(id, partition).second;
@@ -125,7 +137,9 @@ bool UnAckedMessageTrackerEnabled::isEmpty() {
125137

126138
bool UnAckedMessageTrackerEnabled::remove(const MessageId& msgId) {
127139
std::lock_guard<std::recursive_mutex> acquire(lock_);
128-
auto id = discardBatch(msgId);
140+
// Keep consistent with add(): skip discardBatch for ChunkMessageId to ensure
141+
// the same key is used for lookup and removal.
142+
auto id = std::dynamic_pointer_cast<ChunkMessageIdImpl>(msgId.impl_) ? msgId : discardBatch(msgId);
129143
bool removed = false;
130144

131145
std::map<MessageId, std::set<MessageId>&>::iterator exist = messageIdPartitionMap.find(id);

‎tests/MessageChunkingTest.cc‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,17 @@
1818
*/
1919
#include <gtest/gtest.h>
2020
#include <pulsar/Client.h>
21+
#include <pulsar/DeadLetterPolicyBuilder.h>
2122
#include <pulsar/MessageIdBuilder.h>
2223

2324
#include <ctime>
2425
#include <random>
26+
#include <sstream>
2527

2628
#include "PulsarFriend.h"
2729
#include "WaitUtils.h"
2830
#include "lib/ChunkMessageIdImpl.h"
31+
#include "lib/ConsumerImpl.h"
2932
#include "lib/LogUtils.h"
3033

3134
DECLARE_LOG_OBJECT()
@@ -454,6 +457,153 @@ TEST_P(MessageChunkingTest, testResendChunkWithAckHoleMessages) {
454457
consumer.close();
455458
}
456459

460+
// Aligned with Go TestChunkAckAndNAck and Java testNegativeAckChunkedMessage
461+
TEST_P(MessageChunkingTest, testNegativeAckChunkedMessage) {
462+
if (toString(GetParam()) != "None") {
463+
return;
464+
}
465+
const std::string topic =
466+
"MessageChunkingTest-testNegativeAckChunkedMessage-" + std::to_string(time(nullptr));
467+
468+
Consumer consumer;
469+
ConsumerConfiguration consumerConf;
470+
consumerConf.setConsumerType(ConsumerShared);
471+
consumerConf.setNegativeAckRedeliveryDelayMs(1000);
472+
createConsumer(topic, consumer, consumerConf);
473+
474+
Producer producer;
475+
createProducer(topic, producer);
476+
477+
// Send a chunked message
478+
MessageId sendMsgId;
479+
ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMsgId));
480+
481+
// Receive and nack
482+
Message msg;
483+
ASSERT_EQ(ResultOk, consumer.receive(msg, 5000));
484+
ASSERT_EQ(msg.getDataAsString(), largeMessage);
485+
consumer.negativeAcknowledge(msg);
486+
487+
// The message should be redelivered after nack delay
488+
Message redeliveredMsg;
489+
ASSERT_EQ(ResultOk, consumer.receive(redeliveredMsg, 5000));
490+
ASSERT_EQ(redeliveredMsg.getDataAsString(), largeMessage);
491+
consumer.acknowledge(redeliveredMsg);
492+
493+
// Verify no more messages
494+
Message noMsg;
495+
ASSERT_NE(ResultOk, consumer.receive(noMsg, 2000));
496+
497+
producer.close();
498+
consumer.close();
499+
}
500+
501+
// Aligned with Java testLargeMessageAckTimeOut
502+
TEST_P(MessageChunkingTest, testAckTimeoutChunkedMessage) {
503+
if (toString(GetParam()) != "None") {
504+
return;
505+
}
506+
const std::string topic =
507+
"MessageChunkingTest-testAckTimeoutChunkedMessage-" + std::to_string(time(nullptr));
508+
509+
Consumer consumer;
510+
ConsumerConfiguration consumerConf;
511+
consumerConf.setConsumerType(ConsumerShared);
512+
// Set ack timeout to 2 seconds
513+
PulsarFriend::setConsumerUnAckMessagesTimeoutMs(consumerConf, 2000);
514+
createConsumer(topic, consumer, consumerConf);
515+
516+
Producer producer;
517+
createProducer(topic, producer);
518+
519+
// Send a chunked message
520+
MessageId sendMsgId;
521+
ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMsgId));
522+
523+
// Receive but do NOT acknowledge - let ack timeout trigger redelivery
524+
Message msg;
525+
ASSERT_EQ(ResultOk, consumer.receive(msg, 5000));
526+
ASSERT_EQ(msg.getDataAsString(), largeMessage);
527+
528+
// Wait for ack timeout to trigger redelivery
529+
// The message should be redelivered after ack timeout (2s)
530+
Message redeliveredMsg;
531+
ASSERT_EQ(ResultOk, consumer.receive(redeliveredMsg, 5000));
532+
ASSERT_EQ(redeliveredMsg.getDataAsString(), largeMessage);
533+
consumer.acknowledge(redeliveredMsg);
534+
535+
// Verify no more messages
536+
Message noMsg;
537+
ASSERT_NE(ResultOk, consumer.receive(noMsg, 2000));
538+
539+
producer.close();
540+
consumer.close();
541+
}
542+
543+
TEST_P(MessageChunkingTest, testChunkedMessageDLQ) {
544+
if (toString(GetParam()) != "None") {
545+
return;
546+
}
547+
const std::string topic = "persistent://public/default/MessageChunkingTest-testChunkedMessageDLQ-" +
548+
std::to_string(time(nullptr));
549+
const std::string subName = "my-sub";
550+
const std::string dlqTopic = topic + "-" + subName + "-DLQ";
551+
552+
Client client(lookupUrl);
553+
554+
auto dlqPolicy =
555+
DeadLetterPolicyBuilder().maxRedeliverCount(2).initialSubscriptionName("dlq-init-sub").build();
556+
557+
Consumer consumer;
558+
ConsumerConfiguration consumerConf;
559+
consumerConf.setConsumerType(ConsumerShared);
560+
consumerConf.setNegativeAckRedeliveryDelayMs(100);
561+
consumerConf.setDeadLetterPolicy(dlqPolicy);
562+
ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConf, consumer));
563+
564+
// Subscribe to DLQ topic to verify messages arrive there
565+
Consumer dlqConsumer;
566+
ConsumerConfiguration dlqConsumerConf;
567+
dlqConsumerConf.setConsumerType(ConsumerShared);
568+
ASSERT_EQ(ResultOk, client.subscribe(dlqTopic, "dlq-sub", dlqConsumerConf, dlqConsumer));
569+
570+
Producer producer;
571+
createProducer(topic, producer);
572+
573+
// Send a chunked message
574+
MessageId sendMsgId;
575+
ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMsgId));
576+
577+
// Nack the message maxRedeliverCount + 1 times to trigger DLQ
578+
Message msg;
579+
for (int i = 0; i < dlqPolicy.getMaxRedeliverCount() + 1; i++) {
580+
ASSERT_EQ(ResultOk, consumer.receive(msg, 5000));
581+
ASSERT_EQ(msg.getDataAsString(), largeMessage);
582+
consumer.negativeAcknowledge(msg);
583+
}
584+
585+
// Verify the message arrives in DLQ with correct content
586+
Message dlqMsg;
587+
ASSERT_EQ(ResultOk, dlqConsumer.receive(dlqMsg, 10000));
588+
ASSERT_EQ(dlqMsg.getDataAsString(), largeMessage);
589+
std::stringstream expectedOriginMsgId;
590+
expectedOriginMsgId << sendMsgId;
591+
ASSERT_EQ(dlqMsg.getProperty(PROPERTY_ORIGIN_MESSAGE_ID), expectedOriginMsgId.str());
592+
ASSERT_EQ(dlqMsg.getProperty(SYSTEM_PROPERTY_REAL_TOPIC), topic);
593+
594+
// Verify no more messages in DLQ
595+
Message noMsg;
596+
ASSERT_NE(ResultOk, dlqConsumer.receive(noMsg, 2000));
597+
598+
// Verify original consumer has no more messages (message was acked after DLQ send)
599+
ASSERT_NE(ResultOk, consumer.receive(noMsg, 2000));
600+
601+
producer.close();
602+
consumer.close();
603+
dlqConsumer.close();
604+
client.close();
605+
}
606+
457607
// The CI env is Ubuntu 16.04, the gtest-dev version is 1.8.0 that doesn't have INSTANTIATE_TEST_SUITE_P
458608
INSTANTIATE_TEST_CASE_P(Pulsar, MessageChunkingTest,
459609
::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD,

0 commit comments

Comments
 (0)