[Java] [v2.5.2] BatchReceivePolicy not strictly complied by ConsumerBase and MessageImpl

[e-marchand-exensa]

Describe the bug
There are at least two inconsistencies about complying BatchReceivePolicy:
A. BatchReceivePolicy should supports maxNumberOfMessages = 0 and maxNumBytes = 0 if timeout is set but ConsumerBase's constructor issue a warning and reset to default BatchReceivePolicy in this case
B. MessageImpl.canAdd does not comply BatchReceivePolicy, allowing more than maxNumberOfMessages if maxSizeOfMessages is not reached

To Reproduce
A. 1. create a consumer providing BatchReceivePolicy.builder().maxNumMessages(0).maxNumBytes(0).timeout(100,TimeUnit.MILLISECONDS).build()
A. 2. see ConsumerBase's constructor log warning:

BatchReceivePolicy maxNumMessages: 0 or maxNumBytes: 0 is less than 0. Reset to DEFAULT_POLICY. batchReceivePolicy: [...]

B. 1. create a consumer providing BatchReceivePolicy.builder().maxNumMessages(1).maxNumBytes(10*1024*1024).timeout(0,null).build()
B. 2. consumer.batchReceive() may returns a Messages<T> with more than 1 message

Expected behavior
A. ConsumerBase should allow BatchReceivePolicy with maxNumberOfMessages and maxNumBytes set to 0 (or -1) if timeout is set

B. consumer.batchReceive() should comply maxNumberOfMessages and maxNumBytes (not or)

Additional context
In addition to those two issues, there may be an inconsistency between MessageImpl.canAdd and ConsumerBase.hasEnoughMessagesForBatchReceive. Not sure about this one but the code for policy compliance should not be spread among several `classes.```

[e-marchand-exensa]

seems related to (or even fixed by) #6865
[edit] not really, only the part B of this issue.

[codelipenghui]

@e-marchand-exensa I have added the test case for part A #8587. Seems the problem can't be reproduced.

[e-marchand-exensa]

@e-marchand-exensa I have added the test case for part A #8587. Seems the problem can't be reproduced.

@codelipenghui, the issue as reported still exists.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java

Lines 117 to 125 in 4316d96

	} else if (userBatchReceivePolicy.getMaxNumMessages() <= 0 && userBatchReceivePolicy.getMaxNumBytes() <= 0) {
	this.batchReceivePolicy = BatchReceivePolicy.builder()
	.maxNumMessages(BatchReceivePolicy.DEFAULT_POLICY.getMaxNumMessages())
	.maxNumBytes(BatchReceivePolicy.DEFAULT_POLICY.getMaxNumBytes())
	.timeout((int) userBatchReceivePolicy.getTimeoutMs(), TimeUnit.MILLISECONDS)
	.build();
	log.warn("BatchReceivePolicy maxNumMessages: {} or maxNumBytes: {} is less than 0. " +
	"Reset to DEFAULT_POLICY. batchReceivePolicy: {}", userBatchReceivePolicy.getMaxNumMessages(),
	userBatchReceivePolicy.getMaxNumBytes(), this.batchReceivePolicy.toString());