handleMessage method polls buffered message without using it

[ferhatsahinkaya]

  private void handleMessage(AmqpMessageImpl message) {
    boolean schedule = false;
    boolean dispatchNow = false;

    synchronized (this) {
      if (handler != null && demand > 0L && buffered.isEmpty()) {
        if (demand != Long.MAX_VALUE) {
          demand--;
        }
        dispatchNow = true;
      } else if (handler != null && demand > 0L) {
        // Buffered messages present, deliver the oldest of those instead
        buffered.add(message);
        message = buffered.poll();
        if (demand != Long.MAX_VALUE) {
          demand--;
        }

        // Schedule a delivery for the next buffered message
        schedule = true;
      } else {
        // Buffer message until we aren't paused
        buffered.add(message);
      }
    }

    // schedule next delivery if appropriate, after earlier delivery to allow chance to pause etc.
    if (schedule) {
      scheduleBufferedMessageDelivery();
    } else if (dispatchNow) {
      deliverMessageToHandler(message);
    }
  }

Version: 3.9.1

In AmqpReceiverImpl.handleMessage implementation above, when following conditions hold, the oldest message in the buffer is removed without using it.

• !buffered.isEmpty()
• demand > 0
• handler != null

Looks like issue can be fixed with either of the following

• Dispatching message immediately to the handler by setting dispatchNow flag instead of schedule flag
• Scheduling delivery for the next buffered message by setting schedule flag without removing the oldest buffered message and updating demand variable

Probably the preferred way would be dispatching the message immediately as there is no need to wait for dispatching the message later.

[cescoffier]

Do you have a reproducer showing the issue?

And yes, I believe that in this case, we need to deliver the message before setting schedule to true.

[ferhatsahinkaya]

Please find the project reproducing the issue here.

I can locally fix the issue by dispatching the message immediately in the following case:

• !buffered.isEmpty()
• demand > 0
• handler != null

[vietj]

thanks, can you make a PR with a test and a fix that we will review?