feat(version): add maxConsumerTimes to support higher client version (#230)

* feat(version): add maxConsumerTimes to support higher client version

* feat(version): add maxConsumerTimes to support higher client version

* feat(version): add maxConsumerTimes to support higher client version

Author: ShannonDing

include/DefaultMQPushConsumer.h

@@ -111,6 +111,8 @@ class ROCKETMQCLIENT_API DefaultMQPushConsumer : public MQConsumer {
   */
   void setConsumeThreadCount(int threadCount);
   int getConsumeThreadCount() const;
+  void setMaxReconsumeTimes(int maxReconsumeTimes);
+  int getMaxReconsumeTimes() const;
 
   /*
     set pullMsg thread count, default value is cpu cores
@@ -144,6 +146,7 @@ class ROCKETMQCLIENT_API DefaultMQPushConsumer : public MQConsumer {
   MQMessageListener* m_pMessageListener;
   int m_consumeMessageBatchMaxSize;
   int m_maxMsgCacheSize;
+  int m_maxReconsumeTimes = -1;
   boost::asio::io_service m_async_ioService;
   boost::scoped_ptr<boost::thread> m_async_service_thread;
 

src/MQClientAPIImpl.cpp

@@ -836,11 +836,16 @@ void MQClientAPIImpl::consumerSendMessageBack(const string addr,
                                               const string& consumerGroup,
                                               int delayLevel,
                                               int timeoutMillis,
+                                              int maxReconsumeTimes,
                                               const SessionCredentials& sessionCredentials) {
   ConsumerSendMsgBackRequestHeader* pRequestHeader = new ConsumerSendMsgBackRequestHeader();
   pRequestHeader->group = consumerGroup;
   pRequestHeader->offset = msg.getCommitLogOffset();
   pRequestHeader->delayLevel = delayLevel;
+  pRequestHeader->unitMode = false;
+  pRequestHeader->originTopic = msg.getTopic();
+  pRequestHeader->originMsgId = msg.getMsgId();
+  pRequestHeader->maxReconsumeTimes = maxReconsumeTimes;
 
   // string addr = socketAddress2IPPort(msg.getStoreHost());
   RemotingCommand request(CONSUMER_SEND_MSG_BACK, pRequestHeader);

src/MQClientAPIImpl.h

@@ -174,6 +174,7 @@ class MQClientAPIImpl {
                                        const string& consumerGroup,
                                        int delayLevel,
                                        int timeoutMillis,
+                                       int maxReconsumeTimes,
                                        const SessionCredentials& sessionCredentials);
 
   virtual void lockBatchMQ(const string& addr,

src/consumer/ConsumeMessageConcurrentlyService.cpp

@@ -194,6 +194,9 @@ void ConsumeMessageConcurrentlyService::ConsumeRequest(boost::weak_ptr<PullReque
                   msgs[i].getReconsumeTimes());
         if (m_pConsumer->getConsumeType() == CONSUME_PASSIVELY) {
           string brokerName = request->m_messageQueue.getBrokerName();
+          if (m_pConsumer->isUseNameSpaceMode()) {
+            MessageAccessor::withNameSpace(msgs[i], m_pConsumer->getNameSpace());
+          }
           if (!m_pConsumer->sendMessageBack(msgs[i], 0, brokerName)) {
             LOG_WARN("Send message back fail, MQ is:%s, its msgId is:%s, index is:%d, re-consume times is:%d",
                      (request->m_messageQueue).toString().c_str(), msgs[i].getMsgId().c_str(), i,

src/consumer/DefaultMQPushConsumer.cpp

@@ -258,8 +258,8 @@ bool DefaultMQPushConsumer::sendMessageBack(MQMessageExt& msg, int delayLevel, s
   else
     brokerAddr = socketAddress2IPPort(msg.getStoreHost());
   try {
-    getFactory()->getMQClientAPIImpl()->consumerSendMessageBack(brokerAddr, msg, getGroupName(), delayLevel, 3000,
-                                                                getSessionCredentials());
+    getFactory()->getMQClientAPIImpl()->consumerSendMessageBack(brokerAddr, msg, getGroupName(), delayLevel,
+                                                                getMaxReconsumeTimes(), 3000, getSessionCredentials());
   } catch (MQException& e) {
     LOG_ERROR(e.what());
     return false;
@@ -918,6 +918,21 @@ void DefaultMQPushConsumer::setConsumeThreadCount(int threadCount) {
 int DefaultMQPushConsumer::getConsumeThreadCount() const {
   return m_consumeThreadCount;
 }
+void DefaultMQPushConsumer::setMaxReconsumeTimes(int maxReconsumeTimes) {
+  if (maxReconsumeTimes > 0) {
+    m_maxReconsumeTimes = maxReconsumeTimes;
+  } else {
+    LOG_ERROR("set maxReconsumeTimes with invalid value");
+  }
+}
+
+int DefaultMQPushConsumer::getMaxReconsumeTimes() const {
+  if (m_maxReconsumeTimes >= 0) {
+    return m_maxReconsumeTimes;
+  }
+  // return 16 as default;
+  return 16;
+}
 
 void DefaultMQPushConsumer::setPullMsgThreadPoolCount(int threadCount) {
   m_pullMsgThreadPoolNum = threadCount;

src/protocol/CommandHeader.cpp

@@ -492,16 +492,20 @@ void ConsumerSendMsgBackRequestHeader::Encode(Json::Value& outData) {
   outData["group"] = group;
   outData["delayLevel"] = delayLevel;
   outData["offset"] = UtilAll::to_string(offset);
-#ifdef ONS
+  outData["unitMode"] = UtilAll::to_string(unitMode);
   outData["originMsgId"] = originMsgId;
   outData["originTopic"] = originTopic;
-#endif
+  outData["maxReconsumeTimes"] = maxReconsumeTimes;
 }
 
 void ConsumerSendMsgBackRequestHeader::SetDeclaredFieldOfCommandHeader(map<string, string>& requestMap) {
   requestMap.insert(pair<string, string>("group", group));
   requestMap.insert(pair<string, string>("delayLevel", UtilAll::to_string(delayLevel)));
   requestMap.insert(pair<string, string>("offset", UtilAll::to_string(offset)));
+  requestMap.insert(pair<string, string>("unitMode", UtilAll::to_string(unitMode)));
+  requestMap.insert(pair<string, string>("originMsgId", originMsgId));
+  requestMap.insert(pair<string, string>("originTopic", originTopic));
+  requestMap.insert(pair<string, string>("maxReconsumeTimes", UtilAll::to_string(maxReconsumeTimes)));
 }
 //<!***************************************************************************
 void GetConsumerListByGroupResponseBody::Decode(const MemoryBlock* mem, vector<string>& cids) {

src/protocol/CommandHeader.h

@@ -456,6 +456,10 @@ class ConsumerSendMsgBackRequestHeader : public CommandHeader {
   string group;
   int delayLevel;
   int64 offset;
+  bool unitMode = false;
+  string originMsgId;
+  string originTopic;
+  int maxReconsumeTimes = 16;
 };
 
 //<!***************************************************************************

test/src/MQClientAPIImpTest.cpp

@@ -170,6 +170,26 @@ TEST(MQClientAPIImplTest, sendMessage) {
   EXPECT_EQ(result.getMessageQueue().getBrokerName(), "testBroker");
   EXPECT_EQ(result.getMessageQueue().getTopic(), "testTopic");
 }
+
+TEST(MQClientAPIImplTest, consumerSendMessageBack) {
+  SessionCredentials sc;
+  MQMessageExt msg;
+  MockMQClientAPIImpl* impl = MockMQClientAPIImplUtil::GetInstance()->GetGtestMockClientAPIImpl();
+  Mock::AllowLeak(impl);
+  MockTcpRemotingClient* pClient = MockMQClientAPIImplUtil::GetInstance()->GetGtestMockRemotingClient();
+  Mock::AllowLeak(pClient);
+  RemotingCommand* pCommandFailed = new RemotingCommand(SYSTEM_ERROR, nullptr);
+  RemotingCommand* pCommandSuccuss = new RemotingCommand(SUCCESS_VALUE, nullptr);
+  EXPECT_CALL(*pClient, invokeSync(_, _, _))
+      .Times(3)
+      .WillOnce(Return(nullptr))
+      .WillOnce(Return(pCommandFailed))
+      .WillOnce(Return(pCommandSuccuss));
+  EXPECT_ANY_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
+  EXPECT_ANY_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
+  EXPECT_NO_THROW(impl->consumerSendMessageBack("127.0.0.0:10911", msg, "testGroup", 0, 1000, 16, sc));
+}
+
 int main(int argc, char* argv[]) {
   InitGoogleMock(&argc, argv);
   testing::GTEST_FLAG(filter) = "MQClientAPIImplTest.*";

test/src/protocol/CommandHeaderTest.cpp

@@ -67,6 +67,43 @@ using rocketmq::UnregisterClientRequestHeader;
 using rocketmq::UpdateConsumerOffsetRequestHeader;
 using rocketmq::ViewMessageRequestHeader;
 
+TEST(commandHeader, ConsumerSendMsgBackRequestHeader) {
+  string group = "testGroup";
+  int delayLevel = 2;
+  int64 offset = 3027;
+  bool unitMode = true;
+  string originMsgId = "testOriginMsgId";
+  string originTopic = "testTopic";
+  int maxReconsumeTimes = 12;
+  ConsumerSendMsgBackRequestHeader header;
+  header.group = group;
+  header.delayLevel = delayLevel;
+  header.offset = offset;
+  header.unitMode = unitMode;
+  header.originMsgId = originMsgId;
+  header.originTopic = originTopic;
+  header.maxReconsumeTimes = maxReconsumeTimes;
+  map<string, string> requestMap;
+  header.SetDeclaredFieldOfCommandHeader(requestMap);
+  EXPECT_EQ(requestMap["group"], group);
+  EXPECT_EQ(requestMap["delayLevel"], "2");
+  EXPECT_EQ(requestMap["offset"], "3027");
+  EXPECT_EQ(requestMap["unitMode"], "1");
+  EXPECT_EQ(requestMap["originMsgId"], originMsgId);
+  EXPECT_EQ(requestMap["originTopic"], originTopic);
+  EXPECT_EQ(requestMap["maxReconsumeTimes"], "12");
+
+  Value outData;
+  header.Encode(outData);
+  EXPECT_EQ(outData["group"], group);
+  EXPECT_EQ(outData["delayLevel"], 2);
+  EXPECT_EQ(outData["offset"], "3027");
+  EXPECT_EQ(outData["unitMode"], "1");
+  EXPECT_EQ(outData["originMsgId"], originMsgId);
+  EXPECT_EQ(outData["originTopic"], originTopic);
+  EXPECT_EQ(outData["maxReconsumeTimes"], 12);
+}
+
 TEST(commandHeader, GetRouteInfoRequestHeader) {
   GetRouteInfoRequestHeader header("testTopic");
   map<string, string> requestMap;