Fixed accessing MessageImpl after it was enqueued on user queue
@@ -727,8 +727,11 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
}
protected boolean enqueueMessageAndCheckBatchReceive(Message<T> message) {
+ int messageSize = message.size();
if (canEnqueueMessage(message) && incomingMessages.offer(message)) {
- increaseIncomingMessageSize(message);
+ // After we have enqueued the messages on `incomingMessages` queue, we cannot touch the message instance
+ // anymore, since for pooled messages, this instance was possibly already been released and recycled.
+ INCOMING_MESSAGES_SIZE_UPDATER.addAndGet(this, messageSize);
}
return hasEnoughMessagesForBatchReceive();
}
@@ -970,10 +973,6 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
return pendingBatchReceives != null && hasNextBatchReceive();
}
- protected void increaseIncomingMessageSize(final Message<?> message) {
- INCOMING_MESSAGES_SIZE_UPDATER.addAndGet(this, message.size());
- }
-
protected void resetIncomingMessageSize() {
INCOMING_MESSAGES_SIZE_UPDATER.set(this, 0);
}
