Fixed accessing MessageImpl after it was enqueued on user queue

[merlimat]

Motivation

There is an exception that is thrown (and not shown, but that's a different issue) when using a consumer with message pooling enabled and consuming from a partitioned topic.
After a short amount of time the consumption silently stops.

The exception is:

java.lang.NullPointerException
	at org.apache.pulsar.client.impl.MessageImpl.size(MessageImpl.java:398)
	at org.apache.pulsar.client.impl.TopicMessageImpl.size(TopicMessageImpl.java:98)
	at org.apache.pulsar.client.impl.ConsumerBase.increaseIncomingMessageSize(ConsumerBase.java:974)
	at org.apache.pulsar.client.impl.ConsumerBase.enqueueMessageAndCheckBatchReceive(ConsumerBase.java:731)
	at org.apache.pulsar.client.impl.MultiTopicsConsumerImpl.messageReceived(MultiTopicsConsumerImpl.java:301)
	at org.apache.pulsar.client.impl.MultiTopicsConsumerImpl.lambda$receiveMessageFromConsumer$8(MultiTopicsConsumerImpl.java:255)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2144)
	at org.apache.pulsar.client.impl.ConsumerImpl.lambda$internalReceiveAsync$3(ConsumerImpl.java:434)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:830)

The issue is easily reproducible by using pulsar-perf consume over a partitioned topic, since pulsar-perf is using message pooling by default.

The root cause is that we are accessing the message object instance after it was enqueued on the queue where the user will pick it up.

The user is doing something like:

while (true) {
   Message<?> msg = consumer.receive();
   ///
   consumer.acknowledge();
   msg.release();
}

When msg.release() gets called, the object fields are nulled and then the object goes back to the pool. For that, from inside the client lib, we cannot touch that object from the moment it gets put on the queue.

[315157973]

Good job!

[hangc0276]

Great Job!

[eolivelli]

@MMirelli this may be a problem that should make some Fallout tests against current master to fail.
I wonder why we haven't caught this, can you please check?

[eolivelli]

Lgtm

It would be good to have a test case that covers this problem