[fix][client] Remove the dead letter candidate entry using the entry-level message id

[SongOf]

Motivation

ConsumerImpl#possibleSendToDeadLetterTopicMessages is always keyed by the entry-level message
id: receiveIndividualMessagesFromBatch() registers a batch under batchMessage
(a MessageIdImpl), and messageReceived() registers a non-batched message under its own
MessageIdImpl.

processPossibleToDLQ() respects that when it looks the entry up:

deadLetterMessages = possibleSendToDeadLetterTopicMessages.get(MessageIdAdvUtils.discardBatch(messageId));

but not when it deletes it, once the messages have reached the DLQ:

possibleSendToDeadLetterTopicMessages.remove(messageId);

MessageIdAdvUtils#equals/#hashCode include the batch index, so as soon as messageId still
carries one the removal silently matches nothing.

That is reachable on a plain multi-topic / partitioned consumer, with no tracing or custom code
involved: MultiTopicsConsumerImpl adds topicMessage.getMessageId() — a TopicMessageIdImpl
that keeps the batch index — to its unacked-message tracker, unlike ConsumerImpl#trackMessage()
which discards it first. When the ack timeout fires, those ids are routed back to
ConsumerImpl#redeliverUnacknowledgedMessages(Set) and reach processPossibleToDLQ() with the
batch index intact.

The consequences for a consumer using batching + a dead letter policy on a partitioned topic:

• the entry is never dropped — the map only ever gets cleared on reconnect or on close — so it
  grows for the lifetime of the consumer;
• because the entry survives, the next redelivery of any message of that same entry finds it again
  and writes the whole batch to the dead letter topic a second time, so the DLQ accumulates
  duplicates.

Modifications

ConsumerImpl#processPossibleToDLQ(): compute the entry-level key once
(MessageIdAdvUtils.discardBatch(messageId)) and use it for both the lookup and the removal.

acknowledgeAsync(messageId) deliberately keeps using the original id — acknowledging is
per-batch-index and must not have the batch index stripped.

Verifying this change

This change added tests and can be verified as follows:

• DeadLetterTopicTest#testPossibleToDeadLetterEntryIsRemovedWhenRedeliveringABatchMessageId —
  publishes one batch, lets it reach maxRedeliverCount so the batch is registered as a DLQ
  candidate, then redelivers a single message id that still carries its batch index (the shape
  MultiTopicsConsumerImpl passes down), and asserts the entry is gone once the batch has been
  written to the DLQ. The test first asserts that all messages of the batch really did arrive on
  the dead letter topic, so the final assertion cannot pass vacuously. On the unpatched code it
  fails for the real reason:

  org.awaitility.core.ConditionTimeoutException: ... [the entry must be dropped once its batch reached the DLQ]
  Expecting empty but was: {3:0:-1=[MessageImpl@6c435cb6, ...]} within 10 seconds.
  

Local runs:

./gradlew :pulsar-broker:test --tests "DeadLetterTopicTest" \
    --tests "DeadLetterTopicDefaultMultiPartitionsTest" -PtestRetryCount=0
BUILD SUCCESSFUL

./gradlew rat spotlessCheck checkstyleMain checkstyleTest
BUILD SUCCESSFUL

The :pulsar-client-original:test module passes as well.

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

This is an internal bug fix in the consumer's dead letter bookkeeping. It changes no public API,
schema, configuration default, wire protocol, REST endpoint, CLI option or metric.

[SongOf]

@nodece could you review this pr