tests/AcknowledgeTest.cc - pulsar-client-cpp - Git at Google

 /**
  * 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 <pulsar/Client.h>

 #include <atomic>
 #include <chrono>
 #include <set>
 #include <thread>

 #include "ConsumerWrapper.h"
 #include "HttpHelper.h"
 #include "PulsarFriend.h"
 #include "lib/Latch.h"
 #include "lib/LogUtils.h"
 #include "lib/MessageIdUtil.h"

 DECLARE_LOG_OBJECT()

 using namespace pulsar;

 static std::string lookupUrl = "pulsar://localhost:6650";
 static std::string adminUrl = "http://localhost:8080/";

 extern std::string unique_str();

 class AcknowledgeTest
     : public testing::TestWithParam<std::tuple<int /* ack grouping time */, bool /* ack with receipt */>> {};

 TEST_P(AcknowledgeTest, testAckMsgList) {
     Client client(lookupUrl);
     auto clientImplPtr = PulsarFriend::getClientImplPtr(client);

     constexpr auto numMsg = 100;
     std::string uniqueChunk = unique_str();
     std::string topicName = "persistent://public/default/test-ack-msgs" + uniqueChunk;
     const std::string subName = "sub-ack-list";

     Producer producer;
     ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));

     ConsumerConfiguration consumerConfig;
     consumerConfig.setAckGroupingMaxSize(numMsg);
     consumerConfig.setAckGroupingTimeMs(std::get<0>(GetParam()));
     consumerConfig.setAckReceiptEnabled(std::get<1>(GetParam()));
     consumerConfig.setUnAckedMessagesTimeoutMs(10000);
     Consumer consumer;
     ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumerConfig, consumer));

     // Sending and receiving messages.
     for (auto count = 0; count < numMsg; ++count) {
         Message msg = MessageBuilder().setContent(std::string("MSG-") + std::to_string(count)).build();
         ASSERT_EQ(ResultOk, producer.send(msg));
     }

     std::vector<MessageId> recvMsgId;
     for (auto count = 0; count < numMsg; ++count) {
         Message msg;
         ASSERT_EQ(ResultOk, consumer.receive(msg, 1000));
         recvMsgId.emplace_back(msg.getMessageId());
     }
     ASSERT_EQ(ResultOk, consumer.acknowledge(recvMsgId));

     // try redeliver unack messages.
     consumer.redeliverUnacknowledgedMessages();

     auto consumerStats = PulsarFriend::getConsumerStatsPtr(consumer);
     auto ackMap = consumerStats->getAckedMsgMap();
     unsigned long totalAck = ackMap[std::make_pair(ResultOk, CommandAck_AckType_Individual)];
     ASSERT_EQ(totalAck, numMsg);

     Message msg;
     auto ret = consumer.receive(msg, 1000);
     ASSERT_EQ(ResultTimeout, ret) << "Received redundant message ID: " << msg.getMessageId();

     producer.close();
     consumer.close();
     client.close();
 }

 TEST_P(AcknowledgeTest, testAckMsgListWithMultiConsumer) {
     Client client(lookupUrl);
     auto clientImplPtr = PulsarFriend::getClientImplPtr(client);

     std::string uniqueChunk = unique_str();
     std::string topicName = "persistent://public/default/test-ack-msgs" + uniqueChunk;

     // call admin api to make it partitioned
     std::string url =
         adminUrl + "admin/v2/persistent/public/default/test-ack-msgs" + uniqueChunk + "/partitions";
     int res = makePutRequest(url, "5");
     LOG_INFO("res = " << res);
     ASSERT_FALSE(res != 204 && res != 409);

     constexpr auto numMsg = 100;
     const std::string subName = "sub-ack-list";

     Producer producer;
     ProducerConfiguration producerConfig;
     // Turn off batch to ensure even distribution
     producerConfig.setBatchingEnabled(false);
     producerConfig.setPartitionsRoutingMode(pulsar::ProducerConfiguration::RoundRobinDistribution);
     ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfig, producer));

     ConsumerConfiguration consumerConfig;
     // set ack grouping max size is 10
     consumerConfig.setAckGroupingMaxSize(10);
     consumerConfig.setAckGroupingTimeMs(std::get<0>(GetParam()));
     consumerConfig.setAckReceiptEnabled(std::get<1>(GetParam()));
     consumerConfig.setUnAckedMessagesTimeoutMs(10000);
     Consumer consumer;
     ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumerConfig, consumer));

     // Sending and receiving messages.
     for (auto count = 0; count < numMsg; ++count) {
         Message msg = MessageBuilder().setContent(std::string("MSG-") + std::to_string(count)).build();
         ASSERT_EQ(ResultOk, producer.send(msg));
     }

     std::vector<MessageId> recvMsgId;
     for (auto count = 0; count < numMsg; ++count) {
         Message msg;
         ASSERT_EQ(ResultOk, consumer.receive(msg, 1000));
         recvMsgId.emplace_back(msg.getMessageId());
     }
     ASSERT_EQ(ResultOk, consumer.acknowledge(recvMsgId));

     // try redeliver unack messages.
     consumer.redeliverUnacknowledgedMessages();

     // assert stats
     unsigned long totalAck = 0;
     auto consumerStatsList = PulsarFriend::getConsumerStatsPtrList(consumer);
     for (auto consumerStats : consumerStatsList) {
         auto ackMap = consumerStats->getAckedMsgMap();
         totalAck += ackMap[std::make_pair(ResultOk, CommandAck_AckType_Individual)];
     }
     ASSERT_EQ(totalAck, numMsg);

     Message msg;
     auto ret = consumer.receive(msg, 1000);
     ASSERT_EQ(ResultTimeout, ret) << "Received redundant message ID: " << msg.getMessageId();

     producer.close();
     consumer.close();
     client.close();
 }

 TEST_F(AcknowledgeTest, testBatchedMessageId) {
     Client client(lookupUrl);

     const std::string topic = "test-batched-message-id-" + unique_str();
     constexpr int batchingMaxMessages = 3;
     constexpr int numMessages = batchingMaxMessages * 3;

     Producer producer;
     ASSERT_EQ(ResultOk, client.createProducer(topic,
                                               ProducerConfiguration()
                                                   .setBatchingMaxMessages(batchingMaxMessages)
                                                   .setBatchingMaxPublishDelayMs(3600 * 1000 /* 1h */),
                                               producer));
     std::vector<ConsumerWrapper> consumers{4};
     for (size_t i = 0; i < consumers.size(); i++) {
         consumers[i].initialize(client, topic, "sub-" + std::to_string(i));
     }
     for (size_t i = 0; i < numMessages; i++) {
         producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), nullptr);
     }
     for (size_t i = 0; i < consumers.size(); i++) {
         consumers[i].receiveAtMost(numMessages);
         if (i > 0) {
             ASSERT_EQ(consumers[i].messageIdList(), consumers[0].messageIdList());
         }
     }

     Message msg;
     // ack 2 messages of the batch that has 3 messages
     consumers[0].acknowledgeAndRedeliver({0, 2}, AckType::INDIVIDUAL);
     ASSERT_EQ(consumers[0].getConsumer().receive(msg, 1000), ResultOk);
     EXPECT_EQ(msg.getMessageId(), consumers[0].messageIdList()[0]);
     ASSERT_EQ(consumers[0].getNumAcked(CommandAck_AckType_Individual), 0);

     // ack the whole batch
     consumers[1].acknowledgeAndRedeliver({0, 1, 2}, AckType::INDIVIDUAL);
     ASSERT_EQ(consumers[1].getConsumer().receive(msg, 1000), ResultOk);
     EXPECT_EQ(msg.getMessageId(), consumers[1].messageIdList()[batchingMaxMessages]);
     ASSERT_EQ(consumers[1].getNumAcked(CommandAck_AckType_Individual), 3);

     // ack cumulatively the previous message id
     consumers[2].acknowledgeAndRedeliver({batchingMaxMessages, batchingMaxMessages + 1}, AckType::CUMULATIVE);
     ASSERT_EQ(consumers[2].getConsumer().receive(msg, 1000), ResultOk);
     EXPECT_EQ(msg.getMessageId(), consumers[2].messageIdList()[batchingMaxMessages]);
     // the previous message id will only be acknowledged once
     ASSERT_EQ(consumers[2].getNumAcked(CommandAck_AckType_Cumulative), 1);

     // the whole 2nd batch is acknowledged
     consumers[3].acknowledgeAndRedeliver({batchingMaxMessages + 2}, AckType::CUMULATIVE);
     ASSERT_EQ(consumers[3].getConsumer().receive(msg, 1000), ResultOk);
     EXPECT_EQ(msg.getMessageId(), consumers[3].messageIdList()[batchingMaxMessages * 2]);
     ASSERT_EQ(consumers[3].getNumAcked(CommandAck_AckType_Cumulative), 1);
 }

 TEST_F(AcknowledgeTest, testBatchIndexAck) {
     Client client(lookupUrl);
     const std::string topic = "test-batch-index-ack-" + unique_str();
     Producer producer;
     ASSERT_EQ(ResultOk, client.createProducer(
                             topic,
                             ProducerConfiguration().setBatchingMaxMessages(100).setBatchingMaxPublishDelayMs(
                                 3600 * 1000 /* 1h */),
                             producer));
     std::vector<ConsumerWrapper> consumers{3};
     for (size_t i = 0; i < consumers.size(); i++) {
         consumers[i].initialize(client, topic, "sub-" + std::to_string(i), true /* enable batch index ack */);
     }
     constexpr int numMessages = 5;
     for (int i = 0; i < numMessages; i++) {
         producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), nullptr);
     }
     producer.flush();
     for (int i = 0; i < consumers.size(); i++) {
         consumers[i].receiveAtMost(numMessages);
         if (i >= 0) {
             ASSERT_EQ(consumers[0].messageIdList(), consumers[i].messageIdList());
         }
     }
     auto msgIds = consumers[0].messageIdList();

     consumers[0].acknowledgeAndRestart({0, 2, 4}, AckType::INDIVIDUAL);
     consumers[0].receiveAtMost(2);
     ASSERT_EQ(subMessageIdList(msgIds, {1, 3}), consumers[0].messageIdList());

     consumers[1].acknowledgeAndRestart({0, 3}, AckType::INDIVIDUAL_LIST);
     consumers[1].receiveAtMost(3);
     ASSERT_EQ(subMessageIdList(msgIds, {1, 2, 4}), consumers[1].messageIdList());

     consumers[2].acknowledgeAndRestart({3}, AckType::CUMULATIVE);
     consumers[2].receiveAtMost(1);
     ASSERT_EQ(subMessageIdList(msgIds, {4}), consumers[2].messageIdList());
 }

 TEST_F(AcknowledgeTest, testMixedCumulativeAck) {
     Client client(lookupUrl);
     const std::string topic = "test-mixed-cumulative-ack-" + unique_str();

     auto sendBatch = [&](int numMessages) {
         Producer producer;
         auto conf =
             ProducerConfiguration().setBatchingMaxMessages(100).setBatchingMaxPublishDelayMs(3600 * 1000);
         ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer));
         Latch latch{numMessages};
         for (int i = 0; i < numMessages; i++) {
             producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(),
                                [&latch](Result, const MessageId&) { latch.countdown(); });
         }
         producer.flush();
         ASSERT_TRUE(latch.wait(std::chrono::seconds(3)));
     };
     auto sendNonBatch = [&](const std::string& msg) {
         Producer producer;
         auto conf = ProducerConfiguration().setBatchingEnabled(false);
         ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer));
         producer.send(MessageBuilder().setContent(msg).build());
         producer.close();
     };

     ConsumerWrapper consumer;
     consumer.initialize(client, topic, "sub", true);
     sendNonBatch("msg");
     sendBatch(5);
     consumer.receiveAtMost(6);

     // Acknowledge the 1st MessageId, which is a MessageIdImpl
     auto msgIds = consumer.messageIdList();
     ASSERT_TRUE(msgIds[0].batchIndex() < 0);
     consumer.acknowledgeAndRestart({0}, AckType::CUMULATIVE);
     consumer.receiveAtMost(5);
     ASSERT_EQ(subMessageIdList(msgIds, {1, 2, 3, 4, 5}), consumer.messageIdList());

     // Acknowledge the 3rd MessageId, which is a BatchedMessageIdImpl
     ASSERT_EQ(consumer.messageIdList()[2].batchIndex(), 2);
     consumer.acknowledgeAndRestart({2}, AckType::CUMULATIVE);
     consumer.receiveAtMost(2);
     ASSERT_EQ(subMessageIdList(msgIds, {4, 5}), consumer.messageIdList());

     consumer.getConsumer().acknowledgeCumulative(discardBatch(consumer.messageIdList()[0]));
     consumer.getConsumer().close();
     consumer.initialize(client, topic, "sub", true);
     Message msg;
     ASSERT_EQ(ResultTimeout, consumer.getConsumer().receive(msg, 1000));
 }

 TEST_F(AcknowledgeTest, testInvalidMessageId) {
     Client client(lookupUrl);
     std::vector<std::string> topics{"test-invalid-message-id0" + unique_str(),
                                     "test-invalid-message-id1" + unique_str()};
     Consumer consumer;
     ASSERT_EQ(ResultOk, client.subscribe(topics, "sub", consumer));

     Message msg;
     ASSERT_EQ(ResultOperationNotSupported, consumer.acknowledge(msg));
     msg = MessageBuilder().setContent("msg").build();
     ASSERT_EQ(ResultOperationNotSupported, consumer.acknowledge(msg));
 }

 TEST_F(AcknowledgeTest, testAckReceiptEnabled) {
     Client client(lookupUrl);
     const std::string topic = "test-ack-receipt-enabled" + unique_str();

     Producer producer;
     ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
     std::vector<MessageId> msgIds;
     for (int i = 0; i < 5; i++) {
         MessageId msgId;
         ASSERT_EQ(ResultOk,
                   producer.send(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), msgId));
         msgIds.emplace_back(msgId);
     }

     constexpr long ackGroupingTimeMs = 200;
     Consumer consumer;
     ConsumerConfiguration conf;
     conf.setAckGroupingTimeMs(ackGroupingTimeMs);
     conf.setAckReceiptEnabled(true);
     conf.setBatchIndexAckEnabled(true);
     conf.setSubscriptionInitialPosition(InitialPositionEarliest);

     using namespace std::chrono;
     // The ACK grouping timer starts after it's subscribed successfully. To ensure the acknowledgments
     // complete after `ackGroupingTimeMs`, record the start timestamp before subscribing
     auto now = high_resolution_clock::now();
     ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", conf, consumer));

     std::atomic<decltype(std::this_thread::get_id())> threadId[3];
     std::atomic_long durationMs[3];
     std::atomic<Result> result[3];
     Latch latch{3};

     auto createCallback = [&](int i) -> ResultCallback {
         return [i, now, &threadId, &durationMs, &result, &latch](Result result0) {
             threadId[i] = std::this_thread::get_id();
             durationMs[i] =
                 duration_cast<std::chrono::milliseconds>(high_resolution_clock::now() - now).count();
             result[i] = result0;
             latch.countdown();
         };
     };
     consumer.acknowledgeAsync(msgIds[1], createCallback(0));
     consumer.acknowledgeCumulativeAsync(msgIds[2], createCallback(1));
     consumer.acknowledgeAsync(msgIds, createCallback(2));
     ASSERT_TRUE(latch.wait(std::chrono::seconds(3)));
     for (int i = 0; i < 3; i++) {
         LOG_INFO("Ack time: " << durationMs[i] << "ms");
         EXPECT_TRUE(durationMs[i] > ackGroupingTimeMs && durationMs[i] < ackGroupingTimeMs + 100);
         EXPECT_NE(threadId[i], std::this_thread::get_id());
         EXPECT_EQ(result[i], ResultOk);
     }

     client.close();
 }

 INSTANTIATE_TEST_SUITE_P(BasicEndToEndTest, AcknowledgeTest,
                          testing::Combine(testing::Values(100, 0), testing::Values(true, false)),
                          [](const testing::TestParamInfo<std::tuple<int, bool>>& info) {
                              return std::to_string(std::get<0>(info.param)) + "_" +
                                     std::to_string(std::get<1>(info.param));
                          });
	/**
	* 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 <pulsar/Client.h>

	#include <atomic>
	#include <chrono>
	#include <set>
	#include <thread>

	#include "ConsumerWrapper.h"
	#include "HttpHelper.h"
	#include "PulsarFriend.h"
	#include "lib/Latch.h"
	#include "lib/LogUtils.h"
	#include "lib/MessageIdUtil.h"

	DECLARE_LOG_OBJECT()

	using namespace pulsar;

	static std::string lookupUrl = "pulsar://localhost:6650";
	static std::string adminUrl = "http://localhost:8080/";

	extern std::string unique_str();

	class AcknowledgeTest
	: public testing::TestWithParam<std::tuple<int /* ack grouping time /, bool / ack with receipt */>> {};

	TEST_P(AcknowledgeTest, testAckMsgList) {
	Client client(lookupUrl);
	auto clientImplPtr = PulsarFriend::getClientImplPtr(client);

	constexpr auto numMsg = 100;
	std::string uniqueChunk = unique_str();
	std::string topicName = "persistent://public/default/test-ack-msgs" + uniqueChunk;
	const std::string subName = "sub-ack-list";

	Producer producer;
	ASSERT_EQ(ResultOk, client.createProducer(topicName, producer));

	ConsumerConfiguration consumerConfig;
	consumerConfig.setAckGroupingMaxSize(numMsg);
	consumerConfig.setAckGroupingTimeMs(std::get<0>(GetParam()));
	consumerConfig.setAckReceiptEnabled(std::get<1>(GetParam()));
	consumerConfig.setUnAckedMessagesTimeoutMs(10000);
	Consumer consumer;
	ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumerConfig, consumer));

	// Sending and receiving messages.
	for (auto count = 0; count < numMsg; ++count) {
	Message msg = MessageBuilder().setContent(std::string("MSG-") + std::to_string(count)).build();
	ASSERT_EQ(ResultOk, producer.send(msg));
	}

	std::vector<MessageId> recvMsgId;
	for (auto count = 0; count < numMsg; ++count) {
	Message msg;
	ASSERT_EQ(ResultOk, consumer.receive(msg, 1000));
	recvMsgId.emplace_back(msg.getMessageId());
	}
	ASSERT_EQ(ResultOk, consumer.acknowledge(recvMsgId));

	// try redeliver unack messages.
	consumer.redeliverUnacknowledgedMessages();

	auto consumerStats = PulsarFriend::getConsumerStatsPtr(consumer);
	auto ackMap = consumerStats->getAckedMsgMap();
	unsigned long totalAck = ackMap[std::make_pair(ResultOk, CommandAck_AckType_Individual)];
	ASSERT_EQ(totalAck, numMsg);

	Message msg;
	auto ret = consumer.receive(msg, 1000);
	ASSERT_EQ(ResultTimeout, ret) << "Received redundant message ID: " << msg.getMessageId();

	producer.close();
	consumer.close();
	client.close();
	}

	TEST_P(AcknowledgeTest, testAckMsgListWithMultiConsumer) {
	Client client(lookupUrl);
	auto clientImplPtr = PulsarFriend::getClientImplPtr(client);

	std::string uniqueChunk = unique_str();
	std::string topicName = "persistent://public/default/test-ack-msgs" + uniqueChunk;

	// call admin api to make it partitioned
	std::string url =
	adminUrl + "admin/v2/persistent/public/default/test-ack-msgs" + uniqueChunk + "/partitions";
	int res = makePutRequest(url, "5");
	LOG_INFO("res = " << res);
	ASSERT_FALSE(res != 204 && res != 409);

	constexpr auto numMsg = 100;
	const std::string subName = "sub-ack-list";

	Producer producer;
	ProducerConfiguration producerConfig;
	// Turn off batch to ensure even distribution
	producerConfig.setBatchingEnabled(false);
	producerConfig.setPartitionsRoutingMode(pulsar::ProducerConfiguration::RoundRobinDistribution);
	ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfig, producer));

	ConsumerConfiguration consumerConfig;
	// set ack grouping max size is 10
	consumerConfig.setAckGroupingMaxSize(10);
	consumerConfig.setAckGroupingTimeMs(std::get<0>(GetParam()));
	consumerConfig.setAckReceiptEnabled(std::get<1>(GetParam()));
	consumerConfig.setUnAckedMessagesTimeoutMs(10000);
	Consumer consumer;
	ASSERT_EQ(ResultOk, client.subscribe(topicName, subName, consumerConfig, consumer));

	// Sending and receiving messages.
	for (auto count = 0; count < numMsg; ++count) {
	Message msg = MessageBuilder().setContent(std::string("MSG-") + std::to_string(count)).build();
	ASSERT_EQ(ResultOk, producer.send(msg));
	}

	std::vector<MessageId> recvMsgId;
	for (auto count = 0; count < numMsg; ++count) {
	Message msg;
	ASSERT_EQ(ResultOk, consumer.receive(msg, 1000));
	recvMsgId.emplace_back(msg.getMessageId());
	}
	ASSERT_EQ(ResultOk, consumer.acknowledge(recvMsgId));

	// try redeliver unack messages.
	consumer.redeliverUnacknowledgedMessages();

	// assert stats
	unsigned long totalAck = 0;
	auto consumerStatsList = PulsarFriend::getConsumerStatsPtrList(consumer);
	for (auto consumerStats : consumerStatsList) {
	auto ackMap = consumerStats->getAckedMsgMap();
	totalAck += ackMap[std::make_pair(ResultOk, CommandAck_AckType_Individual)];
	}
	ASSERT_EQ(totalAck, numMsg);

	Message msg;
	auto ret = consumer.receive(msg, 1000);
	ASSERT_EQ(ResultTimeout, ret) << "Received redundant message ID: " << msg.getMessageId();

	producer.close();
	consumer.close();
	client.close();
	}

	TEST_F(AcknowledgeTest, testBatchedMessageId) {
	Client client(lookupUrl);

	const std::string topic = "test-batched-message-id-" + unique_str();
	constexpr int batchingMaxMessages = 3;
	constexpr int numMessages = batchingMaxMessages * 3;

	Producer producer;
	ASSERT_EQ(ResultOk, client.createProducer(topic,
	ProducerConfiguration()
	.setBatchingMaxMessages(batchingMaxMessages)
	.setBatchingMaxPublishDelayMs(3600 * 1000 /* 1h */),
	producer));
	std::vector<ConsumerWrapper> consumers{4};
	for (size_t i = 0; i < consumers.size(); i++) {
	consumers[i].initialize(client, topic, "sub-" + std::to_string(i));
	}
	for (size_t i = 0; i < numMessages; i++) {
	producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), nullptr);
	}
	for (size_t i = 0; i < consumers.size(); i++) {
	consumers[i].receiveAtMost(numMessages);
	if (i > 0) {
	ASSERT_EQ(consumers[i].messageIdList(), consumers[0].messageIdList());
	}
	}

	Message msg;
	// ack 2 messages of the batch that has 3 messages
	consumers[0].acknowledgeAndRedeliver({0, 2}, AckType::INDIVIDUAL);
	ASSERT_EQ(consumers[0].getConsumer().receive(msg, 1000), ResultOk);
	EXPECT_EQ(msg.getMessageId(), consumers[0].messageIdList()[0]);
	ASSERT_EQ(consumers[0].getNumAcked(CommandAck_AckType_Individual), 0);

	// ack the whole batch
	consumers[1].acknowledgeAndRedeliver({0, 1, 2}, AckType::INDIVIDUAL);
	ASSERT_EQ(consumers[1].getConsumer().receive(msg, 1000), ResultOk);
	EXPECT_EQ(msg.getMessageId(), consumers[1].messageIdList()[batchingMaxMessages]);
	ASSERT_EQ(consumers[1].getNumAcked(CommandAck_AckType_Individual), 3);

	// ack cumulatively the previous message id
	consumers[2].acknowledgeAndRedeliver({batchingMaxMessages, batchingMaxMessages + 1}, AckType::CUMULATIVE);
	ASSERT_EQ(consumers[2].getConsumer().receive(msg, 1000), ResultOk);
	EXPECT_EQ(msg.getMessageId(), consumers[2].messageIdList()[batchingMaxMessages]);
	// the previous message id will only be acknowledged once
	ASSERT_EQ(consumers[2].getNumAcked(CommandAck_AckType_Cumulative), 1);

	// the whole 2nd batch is acknowledged
	consumers[3].acknowledgeAndRedeliver({batchingMaxMessages + 2}, AckType::CUMULATIVE);
	ASSERT_EQ(consumers[3].getConsumer().receive(msg, 1000), ResultOk);
	EXPECT_EQ(msg.getMessageId(), consumers[3].messageIdList()[batchingMaxMessages * 2]);
	ASSERT_EQ(consumers[3].getNumAcked(CommandAck_AckType_Cumulative), 1);
	}

	TEST_F(AcknowledgeTest, testBatchIndexAck) {
	Client client(lookupUrl);
	const std::string topic = "test-batch-index-ack-" + unique_str();
	Producer producer;
	ASSERT_EQ(ResultOk, client.createProducer(
	topic,
	ProducerConfiguration().setBatchingMaxMessages(100).setBatchingMaxPublishDelayMs(
	3600 * 1000 /* 1h */),
	producer));
	std::vector<ConsumerWrapper> consumers{3};
	for (size_t i = 0; i < consumers.size(); i++) {
	consumers[i].initialize(client, topic, "sub-" + std::to_string(i), true /* enable batch index ack */);
	}
	constexpr int numMessages = 5;
	for (int i = 0; i < numMessages; i++) {
	producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), nullptr);
	}
	producer.flush();
	for (int i = 0; i < consumers.size(); i++) {
	consumers[i].receiveAtMost(numMessages);
	if (i >= 0) {
	ASSERT_EQ(consumers[0].messageIdList(), consumers[i].messageIdList());
	}
	}
	auto msgIds = consumers[0].messageIdList();

	consumers[0].acknowledgeAndRestart({0, 2, 4}, AckType::INDIVIDUAL);
	consumers[0].receiveAtMost(2);
	ASSERT_EQ(subMessageIdList(msgIds, {1, 3}), consumers[0].messageIdList());

	consumers[1].acknowledgeAndRestart({0, 3}, AckType::INDIVIDUAL_LIST);
	consumers[1].receiveAtMost(3);
	ASSERT_EQ(subMessageIdList(msgIds, {1, 2, 4}), consumers[1].messageIdList());

	consumers[2].acknowledgeAndRestart({3}, AckType::CUMULATIVE);
	consumers[2].receiveAtMost(1);
	ASSERT_EQ(subMessageIdList(msgIds, {4}), consumers[2].messageIdList());
	}

	TEST_F(AcknowledgeTest, testMixedCumulativeAck) {
	Client client(lookupUrl);
	const std::string topic = "test-mixed-cumulative-ack-" + unique_str();

	auto sendBatch = [&](int numMessages) {
	Producer producer;
	auto conf =
	ProducerConfiguration().setBatchingMaxMessages(100).setBatchingMaxPublishDelayMs(3600 * 1000);
	ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer));
	Latch latch{numMessages};
	for (int i = 0; i < numMessages; i++) {
	producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(),
	[&latch](Result, const MessageId&) { latch.countdown(); });
	}
	producer.flush();
	ASSERT_TRUE(latch.wait(std::chrono::seconds(3)));
	};
	auto sendNonBatch = [&](const std::string& msg) {
	Producer producer;
	auto conf = ProducerConfiguration().setBatchingEnabled(false);
	ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer));
	producer.send(MessageBuilder().setContent(msg).build());
	producer.close();
	};

	ConsumerWrapper consumer;
	consumer.initialize(client, topic, "sub", true);
	sendNonBatch("msg");
	sendBatch(5);
	consumer.receiveAtMost(6);

	// Acknowledge the 1st MessageId, which is a MessageIdImpl
	auto msgIds = consumer.messageIdList();
	ASSERT_TRUE(msgIds[0].batchIndex() < 0);
	consumer.acknowledgeAndRestart({0}, AckType::CUMULATIVE);
	consumer.receiveAtMost(5);
	ASSERT_EQ(subMessageIdList(msgIds, {1, 2, 3, 4, 5}), consumer.messageIdList());

	// Acknowledge the 3rd MessageId, which is a BatchedMessageIdImpl
	ASSERT_EQ(consumer.messageIdList()[2].batchIndex(), 2);
	consumer.acknowledgeAndRestart({2}, AckType::CUMULATIVE);
	consumer.receiveAtMost(2);
	ASSERT_EQ(subMessageIdList(msgIds, {4, 5}), consumer.messageIdList());

	consumer.getConsumer().acknowledgeCumulative(discardBatch(consumer.messageIdList()[0]));
	consumer.getConsumer().close();
	consumer.initialize(client, topic, "sub", true);
	Message msg;
	ASSERT_EQ(ResultTimeout, consumer.getConsumer().receive(msg, 1000));
	}

	TEST_F(AcknowledgeTest, testInvalidMessageId) {
	Client client(lookupUrl);
	std::vector<std::string> topics{"test-invalid-message-id0" + unique_str(),
	"test-invalid-message-id1" + unique_str()};
	Consumer consumer;
	ASSERT_EQ(ResultOk, client.subscribe(topics, "sub", consumer));

	Message msg;
	ASSERT_EQ(ResultOperationNotSupported, consumer.acknowledge(msg));
	msg = MessageBuilder().setContent("msg").build();
	ASSERT_EQ(ResultOperationNotSupported, consumer.acknowledge(msg));
	}

	TEST_F(AcknowledgeTest, testAckReceiptEnabled) {
	Client client(lookupUrl);
	const std::string topic = "test-ack-receipt-enabled" + unique_str();

	Producer producer;
	ASSERT_EQ(ResultOk, client.createProducer(topic, producer));
	std::vector<MessageId> msgIds;
	for (int i = 0; i < 5; i++) {
	MessageId msgId;
	ASSERT_EQ(ResultOk,
	producer.send(MessageBuilder().setContent("msg-" + std::to_string(i)).build(), msgId));
	msgIds.emplace_back(msgId);
	}

	constexpr long ackGroupingTimeMs = 200;
	Consumer consumer;
	ConsumerConfiguration conf;
	conf.setAckGroupingTimeMs(ackGroupingTimeMs);
	conf.setAckReceiptEnabled(true);
	conf.setBatchIndexAckEnabled(true);
	conf.setSubscriptionInitialPosition(InitialPositionEarliest);

	using namespace std::chrono;
	// The ACK grouping timer starts after it's subscribed successfully. To ensure the acknowledgments
	// complete after `ackGroupingTimeMs`, record the start timestamp before subscribing
	auto now = high_resolution_clock::now();
	ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", conf, consumer));

	std::atomic<decltype(std::this_thread::get_id())> threadId[3];
	std::atomic_long durationMs[3];
	std::atomic<Result> result[3];
	Latch latch{3};

	auto createCallback = [&](int i) -> ResultCallback {
	return [i, now, &threadId, &durationMs, &result, &latch](Result result0) {
	threadId[i] = std::this_thread::get_id();
	durationMs[i] =
	duration_cast<std::chrono::milliseconds>(high_resolution_clock::now() - now).count();
	result[i] = result0;
	latch.countdown();
	};
	};
	consumer.acknowledgeAsync(msgIds[1], createCallback(0));
	consumer.acknowledgeCumulativeAsync(msgIds[2], createCallback(1));
	consumer.acknowledgeAsync(msgIds, createCallback(2));
	ASSERT_TRUE(latch.wait(std::chrono::seconds(3)));
	for (int i = 0; i < 3; i++) {
	LOG_INFO("Ack time: " << durationMs[i] << "ms");
	EXPECT_TRUE(durationMs[i] > ackGroupingTimeMs && durationMs[i] < ackGroupingTimeMs + 100);
	EXPECT_NE(threadId[i], std::this_thread::get_id());
	EXPECT_EQ(result[i], ResultOk);
	}

	client.close();
	}

	INSTANTIATE_TEST_SUITE_P(BasicEndToEndTest, AcknowledgeTest,
	testing::Combine(testing::Values(100, 0), testing::Values(true, false)),
	[](const testing::TestParamInfo<std::tuple<int, bool>>& info) {
	return std::to_string(std::get<0>(info.param)) + "_" +
	std::to_string(std::get<1>(info.param));
	});