Remove log unacked msg.

[Technoboy-]

Motivation

For #13383 fix batch message ack issue, but recently users find that there may exist the unacked-count be negative which will print the log :

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

Lines 932 to 934 in ca64c67

	if (unackedMsgs < 0) {
	log.error("unackedMsgs is : {}, ackedMessages : {}, consumer : {}", unackedMsgs, ackedMessages, consumer);
	}

Then we use the below test to reproduce the case :

    @Test
    public void testNegativeAcks()
            throws Exception {
        String topic = BrokerTestUtil.newUniqueName("testNegativeAcks");

        @Cleanup
        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                .subscriptionName("sub1")
                .subscriptionType(SubscriptionType.Shared)
                .enableBatchIndexAcknowledgment(true)
                .negativeAckRedeliveryDelay(100, TimeUnit.MILLISECONDS)
                .ackTimeout(1000, TimeUnit.MILLISECONDS)
                .subscribe();

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

        Set<String> sentMessages = new HashSet<>();

        final int N = 10000;
        for (int i = 0; i < N; i++) {
            String value = "test-" + i;
            producer.sendAsync(value);
            sentMessages.add(value);
        }
        producer.flush();

        // group1
        for (int i = 0; i < 50; i++) {
            Message<String> msg = consumer.receive();
            if (i % 2 == 0) {
                consumer.acknowledgeAsync(msg);
            } else {
                consumer.negativeAcknowledge(msg);
            }
        }
        // group2
        for (int i = 0; i < 50; i++) {
            Message<String> msg = consumer.receive();
            if (i % 2 == 0) {
                consumer.acknowledgeAsync(msg);
            } else {
                consumer.negativeAcknowledge(msg);
            }
        }
        // group3
        for (int i = 0; i < 10; i++) {
            Message<String> msg = consumer.receive();
            if (i % 2 == 0) {
                consumer.acknowledgeAsync(msg);
            } else {
                consumer.negativeAcknowledge(msg);
            }
        }
        Message<String> receive;
        do{
            receive = consumer.receive(20, TimeUnit.SECONDS);
            consumer.acknowledgeAsync(receive);
        } while (receive != null);
    }

Because when redeliver occurs, ack bitset does not bring back to the broker, we can't calculate how many ack-msg we received. So in this case, the unack-msg count may be negative. But nothing impact.
To reduce user confusion, suggest deleting the log now.
Later we may find a good way to resolve the redelivering case.

Documentation

• no-need-doc

[michaeljmarshall]

@Technoboy- following up here, now that we merged #14288, is this still required? Given that the log line helped find a bug, I am concerned that removing the log line might hide other bugs.

[Technoboy-]

@Technoboy- following up here, now that we merged #14288, is this still required? Given that the log line helped find a bug, I am concerned that removing the log line might hide other bugs.

Yes, good idea. Thanks @michaeljmarshall . I have pushed a new patch #14501