[fixbug] [broker] filter messages in pending ack state.

[thetumbled]

Motivation

Currently, broker do not filter messages in pending ack state when dispatching messages to client. If messages in pending ack state are delivered to client, PulsarClientException$TransactionConflictException will be throwed.
PR #14327 has tried to fix one of cases that broker will dispatch messages in pending ack state.
But, there are other cases that will trigger this problem.
For example, if producer send non batch messages, msg1 and msg2. consumer use txn1 ack msg1, use txn2 ack msg2.
then consumer abort txn2, at this time, we should redeliver only msg2. Because msg1 is in pending ack state.

Modifications

filter non batch messages in pending ack state.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change added tests and can be verified.

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: thetumbled#13

[thetumbled]

PTAL, thanks. @congbobo184

[congbobo184]

There have some problem:

1. use Exclusive, why not use cumulative ack?
2. only check the postion, how batch message can be redeliver?

I don't think it's necessary to fix it, at least for now

[thetumbled]

There have some problem:

1. use Exclusive, why not use cumulative ack?
2. only check the postion, how batch message can be redeliver?

I don't think it's necessary to fix it, at least for now

1. i find that only in Exclusive mode can i recur the problem, it may be have something to do with different dispatcher. In shared mode the test code can not recur the issue.
2. batch messages in pending ack state are filtered by your PR [feature][txn] Fix individual ack batch message with transaction abort redevlier duplicate messages #14327, this PR only filter non batch messages in pending ack state.
3. According to my understanding, PR [feature][txn] Fix individual ack batch message with transaction abort redevlier duplicate messages #14327 try to filter batch messages in pending ack state. for example, producer produce a batch message consisting of two messages msg1 and msg2. consumer use txn1 ack msg1, use txn2 ack msg2, then the consumer abort txn2, at this time, the bitset hold in PendingAckHandleImpl is 01, the bitset hold in cursor is 11. So when broker redeliver msgs to consumer triggered by transaction abortion, broker will send a batch message with bitset 11. so the consumer receive msg1 and msg2 again. but msg1 should be filtered.
4. and, i do not understand what you say in [feature][txn] Fix individual ack batch message with transaction abort redevlier duplicate messages #14327, that is the risk of message lost.

[congbobo184]

1. i find that only in Exclusive mode can i recur the problem, it may be have something to do with different dispatcher. In shared mode the test code can not recur the issue.

exclusive redeliver messages from the cursor markdelete

4. and, i do not understand what you say in [feature][txn] Fix individual ack batch message with txn abort redevlier duplicate messages #14327, that is the risk of message lost.

message dispatch is in order, if messages can be filtered, consumers using cumulative ack with txn will lose the messages which have been filtered.

[thetumbled]

the problem described in this pr have been fixed in #19581, so i close this pr.