[C++] Reduce redeliverMessages when message listener is enabled (#10726)

### Motivation

When consumer's message listener is enabled, application acknowledge processed message in message listener callback, but still tracks the message in `internalListener`,  causes sending redundant redeliverMessages

### Modifications

trackMessage before invoke message listener callback in `internalListener`

Author: saosir

lib/ConsumerImpl.cc

@@ -566,14 +566,15 @@ void ConsumerImpl::internalListener() {
         // This will only happen when the connection got reset and we cleared the queue
         return;
     }
+    trackMessage(msg);
     try {
         consumerStatsBasePtr_->receivedMessage(msg, ResultOk);
         lastDequedMessage_ = Optional<MessageId>::of(msg.getMessageId());
         messageListener_(Consumer(shared_from_this()), msg);
     } catch (const std::exception& e) {
         LOG_ERROR(getName() << "Exception thrown from listener" << e.what());
     }
-    messageProcessed(msg);
+    messageProcessed(msg, false);
 }
 
 Result ConsumerImpl::fetchSingleMessageFromBroker(Message& msg) {
@@ -701,7 +702,7 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
     }
 }
 
-void ConsumerImpl::messageProcessed(Message& msg) {
+void ConsumerImpl::messageProcessed(Message& msg, bool track) {
     Lock lock(mutex_);
     lastDequedMessage_ = Optional<MessageId>::of(msg.getMessageId());
 
@@ -712,7 +713,9 @@ void ConsumerImpl::messageProcessed(Message& msg) {
     }
 
     increaseAvailablePermits(currentCnx);
-    trackMessage(msg);
+    if (track) {
+        trackMessage(msg);
+    }
 }
 
 /**

lib/ConsumerImpl.h

@@ -78,7 +78,7 @@ class ConsumerImpl : public ConsumerImplBase,
     uint64_t getConsumerId();
     void messageReceived(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
                          bool& isChecksumValid, proto::MessageMetadata& msgMetadata, SharedBuffer& payload);
-    void messageProcessed(Message& msg);
+    void messageProcessed(Message& msg, bool track = true);
     inline proto::CommandSubscribe_SubType getSubType();
     inline proto::CommandSubscribe_InitialPosition getInitialPosition();
     void handleUnsubscribe(Result result, ResultCallback callback);