[fix][broker] Fix acked count calculation when acknowledging with transaction

[3pacccccc]

Motivation

This PR fixes an issue with the acked count calculation during individual message acknowledgment with transaction when dealing with messages that no longer exist in the pending acknowledgment list. The problem occurs specifically when acknowledging messages that not exist, which leads to incorrect acked count calculations and potential negative unacked message counts.
you can reproduce in TransactionEndToEndTest like below:

        final String topic = "persistent://" + NAMESPACE1 + "/my-topic";
        String subName = "my-sub";

        Consumer<String> consumer = pulsarClient
                .newConsumer(Schema.STRING)
                .topic(topic)
                .subscriptionName(subName)
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();

        Transaction transaction = getTxn();
        for (int j = 0; j < 10; j++) {
            // mock a message that is not exist.
            MessageIdImpl messageId = new MessageIdImpl(9999L, 9999L, 0);
            try {
                consumer.acknowledgeAsync(messageId, transaction).get();
            } catch (Exception e) {

            }
            System.out.println("unack count:" + getPulsarServiceList().get(0).getBrokerService().getTopic(topic, false)
                    .get().get().getSubscription(subName).getConsumers().get(0).getUnackedMessages());
        }

and you can see console output like:

unack count:-1
unack count:-2
unack count:-3
unack count:-4
unack count:-5
unack count:-6
unack count:-7
unack count:-8
unack count:-9
unack count:-10

Modifications

• Consumer.java:

  • Modified getAckOwnerConsumerAndBatchSize to return batch size 0 for non-existent messages

  • Added check in individualAckWithTransaction to set ackedCount = 0 when batchSize <= 0

  • Improved logging for better debugging of invalid acknowledgment attempts

• MessageIndividualAckTest.java:

  • Added tests for normal and transactional individual acknowledgments

  • Tests verify correct behavior with non-existent messages for both Shared and Key_Shared subscription types

  • Confirms that acked counts are properly calculated (0 for non-existent messages, correct counts for valid messages)

• TransactionEndToEndTest.java:

  • Updated test expectations to reflect that unacked counts should not be reduced when acknowledging already-redelivered messages(this test was introduced in [fix][txn] fix ack with txn compute ackedCount error #17016, which includes when a client later attempts to acknowledge these already-redelivered messages, the system was incorrectly reducing the unacked message count again, leading to double counting. redelivery logic already subtracts the entire batch size from the unacked count1, so subsequent acknowledgments should not further reduce this count)

Verifying this change

• Make sure that the change passes the CI checks.

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: 3pacccccc#33

[3pacccccc]

@lhotari Hi, lari, could you take a look when you have a chance? Thanks.