[Messaging] Cumulative ack may lose message.

[congbobo184]

example:
Transaction abort message1, message2 and then clear incomingMessages.
after incomingMessage cleared the message3 can add in incomingMessages. This time we cumulativeAck message3 with txn will lose message1 and message2.

    @Test
    public void transactionCumulativeAck() throws Exception {
        String topic = NAMESPACE1 + "/txn-timeout";

        @Cleanup
        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                .subscriptionName("test")
                .subscribe();

        @Cleanup
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .topic(topic)
                .sendTimeout(0, TimeUnit.SECONDS)
                .enableBatching(false)
                .producerName("txn-timeout")
                .create();

        producer.send("Hello Pulsar one!");
        producer.send("Hello Pulsar two!");

        Transaction transaction = pulsarClient
                .newTransaction()
                .withTransactionTimeout(3, TimeUnit.SECONDS)
                .build().get();

        Message<String> message = consumer.receive();

        System.out.println(message.getValue());

        consumer.acknowledgeCumulativeAsync(message.getMessageId(), transaction).get();

        transaction.abort();

        transaction = pulsarClient
                .newTransaction()
                .withTransactionTimeout(3, TimeUnit.SECONDS)
                .build().get();
        // it may receive Hello Pulsar two!
        Message<String> reReceiveMessage = consumer.receive(2, TimeUnit.SECONDS);
        System.out.println(reReceiveMessage.getValue());

    }

It not a transaction bug, now normal cumulative ack will produce race condition.

consumer pull request and clear incomming queue is not synchronized

when clear completed, we don't send redeliverCommond to server, this time consumer receive MessageCommand the incomming queue will add new message, the messageId may more than the old messageId whtich you don't cumulative ack. then you cumulative ack this new messageId, the old messageID will lost.

may we should add an epoch for every messageCommand, when invoke redeliverUnacknowledgedMessages, we can add the epoch and refuse the messageCommand which is smaller than current consumer epoch.

[codelipenghui]

The issue had no activity for 30 days, mark with Stale label.

[congbobo184]

#10478 fixed