jiazhai commented on issue #2574: Timeout message not get redeliver in 
TopicsConsumer when use message listener
   Penghui meet this issue in their test.
   This issue may be caused by the un-acked messages tracking in both 
individual ConsumerImpl and MultiTopicsConsumerImpl.
   In MultiTopicsConsumerImpl, we are managing and tracking all un-acked 
message.  But when using MessageListener, there will be some issues.  
   This is because MessageListener.received() is only called 1 time when the 
message is delivered from underline individual ConsumerImpl.
   The calling stack is like this:
   MultiTopicsConsumerImpl.startReceivingMessages() // try to receive messages 
from all individual ConsumerImpl
   MultiTopicsConsumerImpl.receiveMessageFromConsumer(ConsumerImpl consumer)
   // call individual ConsumerImpl.receiveAsync() to import messages. And call 
MessageListener.received() in receiveAsync() call back.
   MultiTopicsConsumerImpl.messageReceived()  // called in 
ConsumerImpl.receiveAsync() callback.
   While in ConsumerImpl.messageProcessed(), it not tracking messages for 
   if (partitionIndex != -1) {
       // we should no longer track this message, TopicsConsumer will take care 
from now onwards
       unAckedMessageTracker.remove(id);    < === we are not tracking messages
   } else {
   So for a timed-out message, it will not be redelivered in ConsumerImpl, and 
as above calling stack, MessageListener.received()  will not get called.

This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:

With regards,
Apache Git Services

Reply via email to