[fix][client] Fix internal receive used wrong timeout type

[Demogorgon314]

Fixes #15001

Motivation

Currently, when we called redeliverUnacknowledgedMessages and consumer use receive with timeout unit is second, the timeout might not work.

When we use the second as receive timeout unit like consumer.receive(3, TimeUnit.SECONDS);, line 473 will pass a negative timeout to the next internalReceive calls, because we are subtracting nanoseconds from seconds.

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

Lines 459 to 476 in f3b87b6

	protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
	Message<T> message;
	long callTime = System.nanoTime();
	try {
	message = incomingMessages.poll(timeout, unit);
	if (message == null) {
	return null;
	}
	messageProcessed(message);
	if (!isValidConsumerEpoch(message)) {
	long executionTime = System.nanoTime() - callTime;
	if (executionTime >= unit.toNanos(timeout)) {
	return null;
	} else {
	return internalReceive((int) (timeout - executionTime), TimeUnit.NANOSECONDS);
	}
	}
	return beforeConsume(message);

Modifications

• Use the same type to subtract.
• Use long type as timeout argument.

Documentation

Check the box below or label this PR directly (if you have committer privilege).

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

This is a bug fix, no need for docs.

• doc

  (If this PR contains doc changes)

[Jason918]

Good catch.

[mattisonchao]

After discussing with @Demogorgon314, We don't need to cherry-pick this PR to branch-2.9.
Because the issue involves by #10478