[fix][client] Fix negative ack not redelivery.

[Technoboy-]

Motivation

When consumer subscribes to more than one topic, if uses negative API to redelivery msg, it will fail with the below error:

java.lang.IllegalArgumentException: null
	at com.google.common.base.Preconditions.checkArgument(Preconditions.java:131) ~[guava-31.0.1-jre.jar:?]
	at org.apache.pulsar.client.impl.ConsumerImpl.redeliverUnacknowledgedMessages(ConsumerImpl.java:1901) ~[classes/:?]
	at org.apache.pulsar.client.impl.NegativeAcksTracker.triggerRedelivery(NegativeAcksTracker.java:82) ~[classes/:?]

Because if subscribes with multi-topics, the internal message will be TopicMessageIdImpl and add to NegativeAcksTracker without conversion to MessageIdImpl(user not config with negativeAckRedeliveryBackoff):

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

Lines 87 to 112 in 35c4b68

	public synchronized void add(MessageId messageId) {
	if (messageId instanceof BatchMessageIdImpl) {
	BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId;
	messageId = new MessageIdImpl(batchMessageId.getLedgerId(), batchMessageId.getEntryId(),
	batchMessageId.getPartitionIndex());
	}
	if (nackedMessages == null) {
	nackedMessages = new HashMap<>();
	}
	nackedMessages.put(messageId, System.nanoTime() + nackDelayNanos);
	if (this.timeout == null) {
	// Schedule a task and group all the redeliveries for same period. Leave a small buffer to allow for
	// nack immediately following the current one will be batched into the same redeliver request.
	this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS);
	}
	}
	public synchronized void add(Message<?> message) {
	if (negativeAckRedeliveryBackoff == null) {
	add(message.getMessageId());
	return;
	}
	add(message.getMessageId(), message.getRedeliveryCount());
	}

Then NegativeAcksTracker will trigger redeliver and throw IllegalArgumentException at here :

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

Lines 1896 to 1902 in 35c4b68

	public void redeliverUnacknowledgedMessages(Set<MessageId> messageIds) {
	if (messageIds.isEmpty()) {
	return;
	}
	checkArgument(messageIds.stream().findFirst().get() instanceof MessageIdImpl);

The issue seems introduced by #12566.

Documentation

• no-need-doc
  (Please explain why)

Reproduce Test

@Test
    @SneakyThrows
    public void testNegativeAcks() {
        String topic = BrokerTestUtil.newUniqueName("testNegativeAcks");
        @Cleanup
        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                .subscriptionName("sub1")
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true)
                .negativeAckRedeliveryDelay(1, TimeUnit.SECONDS)
                .subscribe();

        @Cleanup
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .topic(topic)
                .create();

        producer.sendAsync("test-ack");
        producer.flush();
        while(true) {
            Message<String> msg1 = consumer.receive();
            System.out.println("received : " + msg1);
            consumer.negativeAcknowledge(msg1);
        }
    }

[lhotari]

@Technoboy- can you check #6869 , does this PR fix that issue?

[Technoboy-]

@Technoboy- can you check #6869 , does this PR fix that issue?

It's not related to that issue. This bug was introduced by #12566.

[lhotari]

It's not related to that issue. This bug was introduced by #12566.

thanks. btw. would you be interested in fixing #6869 ? That's a long time issue with negative acknowledgements and batch messages.

[lhotari]

LGTM

[Technoboy-]

It's not related to that issue. This bug was introduced by #12566.

thanks. btw. would you be interested in fixing #6869 ? That's a long time issue with negative acknowledgements and batch messages.

I will work on it.

[Technoboy-]

It's not related to that issue. This bug was introduced by #12566.

thanks. btw. would you be interested in fixing #6869 ? That's a long time issue with negative acknowledgements and batch messages.

Hi @lhotari , I think #6869 has been fixed by #9440.