[fix][client] Messages with inconsistent consumer epochs are not filtered when using batch receive and trigger timeout.

[shibd]

Motivation

Messages with inconsistent consumer epochs are not filtered when using batch receive and trigger timeout.

Modifications

• Add consumer epoch on notifyPendingBatchReceivedCallBack.
• Reuse notifyPendingBatchReceivedCallBack logic.

Verifying this change

• testBatchReceiveRedeliveryAddEpoch unit test covers the scene.

Documentation

• doc-not-needed

Matching PR in forked repository

shibd#16

[shibd]

/pulsarbot run-failure-checks

[shibd]

@nodece Thanks, all comments fixed, PTAL.

[shibd]

/pulsarbot run-failure-checks

[codelipenghui]

@congbobo184 Do you remember why the consumer epoch check happened after polling the messages from the receiver queue?

@shibd @congbobo184
Can we move it to the beginning of message receiving

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

Lines 1271 to 1285 in 69f3f74

	void messageReceived(CommandMessage cmdMessage, ByteBuf headersAndPayload, ClientCnx cnx) {
	List<Long> ackSet = Collections.emptyList();
	if (cmdMessage.getAckSetsCount() > 0) {
	ackSet = new ArrayList<>(cmdMessage.getAckSetsCount());
	for (int i = 0; i < cmdMessage.getAckSetsCount(); i++) {
	ackSet.add(cmdMessage.getAckSetAt(i));
	}
	}
	int redeliveryCount = cmdMessage.getRedeliveryCount();
	MessageIdData messageId = cmdMessage.getMessageId();
	long consumerEpoch = DEFAULT_CONSUMER_EPOCH;
	// if broker send messages to client with consumerEpoch, we should set consumerEpoch to message
	if (cmdMessage.hasConsumerEpoch()) {
	consumerEpoch = cmdMessage.getConsumerEpoch();
	}

It looks like if the message with invalid epoch, we don't need to add it to the receiver queue.
And we will also introduce client-side stats issues, because users haven't seen that message, but from stats they see the received messages are increased.

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

Lines 1628 to 1644 in 69f3f74

	protected synchronized void messageProcessed(Message<?> msg) {
	ClientCnx currentCnx = cnx();
	ClientCnx msgCnx = ((MessageImpl<?>) msg).getCnx();
	lastDequeuedMessageId = msg.getMessageId();
	if (msgCnx != currentCnx) {
	// The processed message did belong to the old queue that was cleared after reconnection.
	} else {
	if (listener == null && !parentConsumerHasListener) {
	increaseAvailablePermits(currentCnx);
	}
	stats.updateNumMsgsReceived(msg);
	trackMessage(msg);
	}
	decreaseIncomingMessageSize(msg);
	}

The messages with invalid epoch should be the skipped messages like here

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

Line 1559 in 69f3f74

int skippedMessages = 0;

It also makes sense to add skipped messages in ConsumerStats

[codelipenghui]

Looks good to me.

@shibd I have left some minor comment, PTAL.

[shibd]

/pulsarbot run-failure-checks