[fix][client] Clean up unacked messages when unsubscribing a topic with ack timeout backoff

[Dream95]

Motivation

When a multi-topic consumer unsubscribes from a single topic, the client should drop that topic's unacked messages from the ack-timeout tracker so they are not redelivered later.
MultiTopicsConsumerImpl already called removeTopicMessages for UnAckedTopicMessageTracker, but not for UnAckedTopicMessageRedeliveryTracker, which is used when ackTimeoutRedeliveryBackoff is configured . Unsubscribed topics could therefore leave stale entries in the redelivery partition map and in ackTimeoutMessages.
Additionally, UnAckedTopicMessageRedeliveryTracker.removeTopicMessages used the partition-map iterator when iterating ackTimeoutMessages (iterator.hasNext() / iterator.remove() instead of iteratorAckTimeOut), so ack-timeout entries were not removed correctly and internal tracker state could be corrupted.

Modifications

• Call removeTopicMessages on UnAckedTopicMessageRedeliveryTracker when a topic is unsubscribed from a multi-topic consumer.
• Fix ack-timeout iteration in UnAckedTopicMessageRedeliveryTracker.removeTopicMessages to use iteratorAckTimeOut.
• Add UnAckedTopicMessageRedeliveryTrackerTest to verify removeTopicMessages clears both the partition map and ack-timeout map.

Verifying this change

• ./gradlew :pulsar-client-original:test --tests org.apache.pulsar.client.impl.UnAckedTopicMessageRedeliveryTrackerTest

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Matching PR in forked repository

PR in forked repository: Dream95#12

[lhotari]

LGTM, good work @Dream95

[void-ptr974]

LGTM. Thanks for the fix!

The missing cleanup for UnAckedTopicMessageRedeliveryTracker looks correct, and the iterator fix in removeTopicMessages is important.

One non-blocking note: the current getOwnerTopic().contains(topicName) matching seems a bit loose and may match similar topic names, such as my-topic and my-topic-v2. This is pre-existing behavior, so it shouldn’t block this PR.

Would you be interested in addressing this in a follow-up PR?

[Dream95]

LGTM. Thanks for the fix!

The missing cleanup for UnAckedTopicMessageRedeliveryTracker looks correct, and the iterator fix in removeTopicMessages is important.

One non-blocking note: the current getOwnerTopic().contains(topicName) matching seems a bit loose and may match similar topic names, such as my-topic and my-topic-v2. This is pre-existing behavior, so it shouldn’t block this PR.

Would you be interested in addressing this in a follow-up PR?

Thanks for the review. I'll fix that in a follow-up PR.