[fix] Consumer batch receive will cause data loss.

[shibd]

Motivation

Currently, the batch receive method may cause data loss.

The root cause is, when peekMsg can't add to messages(The batch has satisfied the batch receive policy), it has actually been popped.

pulsar-client-cpp/lib/ConsumerImpl.cc

Lines 652 to 657 in ba7f59d

	Message peekMsg;
	while (incomingMessages_.pop(peekMsg, std::chrono::milliseconds(0)) && messages->canAdd(peekMsg)) {
	messageProcessed(peekMsg);
	Message interceptMsg = interceptors_->beforeConsume(Consumer(shared_from_this()), peekMsg);
	messages->add(interceptMsg);
	}

Modifications

• Peek first, pop only when it can be added.

Verifying this change

• Change testBatchReceive test to cover this case:
  Send 100 messages first, batch receive 10 messages at a time, and assert that all messages can be received.

Documentation

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[BewareMyPower]

Should we also check isClosedNoMutex for both BlockingQueue and UnboundedBlockingQueue?

[shibd]

Should we also check isClosedNoMutex for both BlockingQueue and UnboundedBlockingQueue?

oh, I missed it. added check, PTAL.