[fix][client] Avoid redelivering duplicated messages when batching is enabled by BewareMyPower · Pull Request #18486 · apache/pulsar

BewareMyPower · 2022-11-15T19:55:08Z

Motivation

#18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the testConsumerDedup test modified by #18454.

The root cause is that single messages will still be passed into the isDuplicated method in receiveIndividualMessagesFromBatch. However, in this case, the MessageId is a BatchedMessageIdImpl, while the MessageId in lastCumulativeAck or pendingIndividualAcks are MessageIdImpl implementations.

Modifications

Validate the class type in isDuplicated and convert a BatchedMessageIdImpl to MessageIdImpl. Then revert the unnecessary changes in #18454.

ConsumerRedeliveryTest#testAckNotSent is added to verify it works.

TODO

The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.

Documentation

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

Matching PR in forked repository

PR in forked repository: BewareMyPower#8

### Motivation apache#18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by apache#18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. ### Modifications Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in apache#18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. ### TODO The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.

codelipenghui · 2022-11-16T01:12:24Z

/pulsarbot run-failure-checks

codecov-commenter · 2022-11-16T01:28:36Z

Codecov Report

Merging #18486 (c48ab86) into master (aeb4503) will increase coverage by 16.86%.
The diff coverage is 62.26%.

@@              Coverage Diff              @@
##             master   #18486       +/-   ##
=============================================
+ Coverage     31.39%   48.26%   +16.86%     
- Complexity     6651     9447     +2796     
=============================================
  Files           697      618       -79     
  Lines         68015    58558     -9457     
  Branches       7285     6091     -1194     
=============================================
+ Hits          21353    28262     +6909     
+ Misses        43667    27278    -16389     
- Partials       2995     3018       +23

Flag	Coverage Δ
unittests	`48.26% <62.26%> (+16.86%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...ersistentStreamingDispatcherMultipleConsumers.java	`0.00% <0.00%> (ø)`
...he/pulsar/client/impl/MultiTopicsConsumerImpl.java	`22.98% <0.00%> (+0.71%)`	⬆️
...ersistentStickyKeyDispatcherMultipleConsumers.java	`58.91% <50.00%> (+20.44%)`	⬆️
...sistent/PersistentDispatcherMultipleConsumers.java	`58.13% <75.00%> (+18.90%)`	⬆️
...impl/PersistentAcknowledgmentsGroupingTracker.java	`58.89% <77.27%> (+1.16%)`	⬆️
...pulsar/broker/admin/impl/PersistentTopicsBase.java	`59.22% <100.00%> (+43.45%)`	⬆️
...lsar/client/impl/conf/ClientConfigurationData.java	`96.66% <100.00%> (+0.11%)`	⬆️
...apache/pulsar/client/impl/AutoClusterFailover.java	`75.55% <0.00%> (-0.56%)`	⬇️
...va/org/apache/pulsar/client/impl/ConsumerBase.java	`21.93% <0.00%> (-0.02%)`	⬇️
.../main/java/org/apache/pulsar/PulsarStandalone.java	`0.00% <0.00%> (ø)`
... and 252 more

there are new changes

congbobo184

LGTM! should we add a test
5 messages in one batch and ackIndividual the 3 messages, redeliver, only can receive the last 2 messages in this batch and enableBatchIndex

congbobo184 · 2022-11-18T13:09:19Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java

+                { 3, 5, CommandAck.AckType.Individual },
+                { 5, 5, CommandAck.AckType.Individual },
+                { 3, 5, CommandAck.AckType.Cumulative },
+                { 5, 5, CommandAck.AckType.Cumulative }


5, 5 test case may not stable

I see, I will take a look soon.

It's fixed now. The root cause is that after b0fad40, a BatchedMessageIdImpl will be compared with a MessageIdImpl, which should be forbidden.

@congbobo184 PTAL again.

liangyepianzhou

LGTM, good work, and the root cause of the flaky test should be this.

BewareMyPower · 2022-11-21T02:20:51Z

5 messages in one batch and ackIndividual the 3 messages, redeliver, only can receive the last 2 messages in this batch and enableBatchIndex

@congbobo184 See the TODO in my PR description. I think it's a bug for batch index ACK and it needs some extra work to fix.

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

congbobo184 · 2022-11-21T09:50:27Z

5 messages in one batch and ackIndividual the 3 messages, redeliver, only can receive the last 2 messages in this batch and enableBatchIndex

@congbobo184 See the TODO in my PR description. I think it's a bug for batch index ACK and it needs some extra work to fix.

sorry, I miss your description. doesn't the batch index ack fix should include in release/2.9.4?

…#18486) #18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by #18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in #18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  PR in forked repository: BewareMyPower#8 (cherry picked from commit be1d07e)

BewareMyPower · 2022-11-22T07:32:15Z

doesn't the batch index ack fix should include in release/2.9.4?

I think not. It's not a regression. It's a bug from the beginning when batch index ACK was introduced.

## Motivation 1. fix flaky test #18466 caused by txn async send method 2. decrease run time by optimizing receive method ## Modification 1. fix flaky test * modify `producer.newMessage(txn1).value(("Hello Txn - " + i).getBytes(UTF_8)).sendAsync();` to `producer.newMessage(txn1).value(("Hello Txn - " + i).getBytes(UTF_8)).send();` This also can be resolved by #17836 and #18486 later. 2. decrease run time by optimizing receive method * modify ` Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS); Assert.assertNull(message);` to ` Message<byte[]> message = consumer.receive(300, TimeUnit.MILLISECONDS); Assert.assertNull(message);` * modify `message = consumer.receive();` to `message = consumer.receive(5, TimeUnit.SECONDS);` * keep other `consumer.receive(x, y)` no change.

…#18486) #18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by #18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in #18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  PR in forked repository: BewareMyPower#8 (cherry picked from commit be1d07e)

…apache#18486) apache#18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by apache#18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in apache#18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  PR in forked repository: BewareMyPower#8 (cherry picked from commit be1d07e) (cherry picked from commit 870b060)

…#18486) #18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by #18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in #18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  PR in forked repository: BewareMyPower#8 (cherry picked from commit be1d07e)

…apache#18486) ### Motivation apache#18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by apache#18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. ### Modifications Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in apache#18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. ### TODO The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it. ### Documentation  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  ### Matching PR in forked repository PR in forked repository: BewareMyPower#8

## Motivation 1. fix flaky test apache#18466 caused by txn async send method 2. decrease run time by optimizing receive method ## Modification 1. fix flaky test * modify `producer.newMessage(txn1).value(("Hello Txn - " + i).getBytes(UTF_8)).sendAsync();` to `producer.newMessage(txn1).value(("Hello Txn - " + i).getBytes(UTF_8)).send();` This also can be resolved by apache#17836 and apache#18486 later. 2. decrease run time by optimizing receive method * modify ` Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS); Assert.assertNull(message);` to ` Message<byte[]> message = consumer.receive(300, TimeUnit.MILLISECONDS); Assert.assertNull(message);` * modify `message = consumer.receive();` to `message = consumer.receive(5, TimeUnit.SECONDS);` * keep other `consumer.receive(x, y)` no change.

…apache#18486) ### Motivation apache#18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by apache#18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. ### Modifications Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in apache#18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. ### TODO The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it. ### Documentation  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  ### Matching PR in forked repository PR in forked repository: BewareMyPower#8

## Motivation 1. fix flaky test apache#18466 caused by txn async send method 2. decrease run time by optimizing receive method ## Modification 1. fix flaky test * modify `producer.newMessage(txn1).value(("Hello Txn - " + i).getBytes(UTF_8)).sendAsync();` to `producer.newMessage(txn1).value(("Hello Txn - " + i).getBytes(UTF_8)).send();` This also can be resolved by apache#17836 and apache#18486 later. 2. decrease run time by optimizing receive method * modify ` Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS); Assert.assertNull(message);` to ` Message<byte[]> message = consumer.receive(300, TimeUnit.MILLISECONDS); Assert.assertNull(message);` * modify `message = consumer.receive();` to `message = consumer.receive(5, TimeUnit.SECONDS);` * keep other `consumer.receive(x, y)` no change.

…#18486) #18454 fixed the potential message loss when a batched message is redelivered and one single message of the batch is added to the ACK tracker. However, it also leads to a potential message duplication, see the `testConsumerDedup` test modified by #18454. The root cause is that single messages will still be passed into the `isDuplicated` method in `receiveIndividualMessagesFromBatch`. However, in this case, the `MessageId` is a `BatchedMessageIdImpl`, while the `MessageId` in `lastCumulativeAck` or `pendingIndividualAcks` are `MessageIdImpl` implementations. Validate the class type in `isDuplicated` and convert a `BatchedMessageIdImpl` to `MessageIdImpl`. Then revert the unnecessary changes in #18454. `ConsumerRedeliveryTest#testAckNotSent` is added to verify it works. The duplication could still happen when batch index ACK is enabled. Because even after the ACK tracker is flushed, if only parts of a batched message are not acknowledged, the whole batched message would still be redelivered. I will open another PR to fix it.  - [ ] `doc`  - [ ] `doc-required`  - [x] `doc-not-needed`  - [ ] `doc-complete`  PR in forked repository: BewareMyPower#8

BewareMyPower added type/bug The PR fixed a bug or issue reported a bug release/2.9.4 release/2.8.5 release/2.11.1 release/2.10.3 labels Nov 15, 2022

BewareMyPower requested review from Technoboy-, codelipenghui, congbobo184, eolivelli, mattisonchao and michaeljmarshall November 15, 2022 19:55

BewareMyPower self-assigned this Nov 15, 2022

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Nov 15, 2022

codelipenghui previously approved these changes Nov 16, 2022

View reviewed changes

codelipenghui added this to the 2.12.0 milestone Nov 16, 2022

Fix cumulative ack test

b0fad40

Merge branch 'master' into bewaremypower/fix-duplicated-individual-ack

6b339e1

congbobo184 reviewed Nov 18, 2022

View reviewed changes

eolivelli approved these changes Nov 18, 2022

View reviewed changes

congbobo184 reviewed Nov 18, 2022

View reviewed changes

liangyepianzhou approved these changes Nov 20, 2022

View reviewed changes

This was referenced Nov 20, 2022

Flaky-test: TransactionEndToEndTest.produceCommitTest #18466

Closed

[improve][test] Optimize TransactionEndToEndTest #18522

Merged

BewareMyPower added 2 commits November 21, 2022 10:21

Fix testAckNotSent

321785f

Fix testFailoverConsumerBatchCumulateAck

2561706

Add LastCumulativeAck#compareTo

412bb92

BewareMyPower mentioned this pull request Nov 21, 2022

[fix][client] Fix unable to send cumulative acknowledgment for last messageId of a batch when enable batch index acknowledgment #18531

Closed

5 tasks

BewareMyPower added 2 commits November 21, 2022 12:56

Add tests for LastCumulativeAck#compareTo

d79c502

Fix wrong semantics of compareTo

fc4de9c

coderzc reviewed Nov 21, 2022

View reviewed changes

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java Outdated Show resolved Hide resolved

Enhance compareTo

c48ab86

coderzc approved these changes Nov 21, 2022

View reviewed changes

congbobo184 merged commit be1d07e into apache:master Nov 21, 2022

congbobo184 added the cherry-picked/branch-2.9 Archived: 2.9 is end of life label Nov 21, 2022

BewareMyPower deleted the bewaremypower/fix-duplicated-individual-ack branch November 24, 2022 07:02

Technoboy- changed the title ~~[fix] Avoid redelivering duplicated messages when batching is enabled~~ [fix][client] Avoid redelivering duplicated messages when batching is enabled Nov 28, 2022

liangyepianzhou added the cherry-picked/branch-2.10 label Dec 5, 2022

Technoboy- added the cherry-picked/branch-2.11 label Feb 8, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][client] Avoid redelivering duplicated messages when batching is enabled#18486

[fix][client] Avoid redelivering duplicated messages when batching is enabled#18486
congbobo184 merged 9 commits intoapache:masterfrom
BewareMyPower:bewaremypower/fix-duplicated-individual-ack

BewareMyPower commented Nov 15, 2022

Uh oh!

codelipenghui commented Nov 16, 2022

Uh oh!

codecov-commenter commented Nov 16, 2022 •

edited

Loading

Uh oh!

congbobo184 left a comment •

edited

Loading

Uh oh!

congbobo184 Nov 18, 2022

Uh oh!

BewareMyPower Nov 19, 2022

Uh oh!

BewareMyPower Nov 21, 2022

Uh oh!

BewareMyPower Nov 21, 2022

Uh oh!

liangyepianzhou left a comment

Uh oh!

BewareMyPower commented Nov 21, 2022

Uh oh!

Uh oh!

congbobo184 commented Nov 21, 2022

Uh oh!

BewareMyPower commented Nov 22, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

Conversation

BewareMyPower commented Nov 15, 2022

Motivation

Modifications

TODO

Documentation

Matching PR in forked repository

Uh oh!

codelipenghui commented Nov 16, 2022

Uh oh!

codecov-commenter commented Nov 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

congbobo184 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 Nov 18, 2022

Choose a reason for hiding this comment

Uh oh!

BewareMyPower Nov 19, 2022

Choose a reason for hiding this comment

Uh oh!

BewareMyPower Nov 21, 2022

Choose a reason for hiding this comment

Uh oh!

BewareMyPower Nov 21, 2022

Choose a reason for hiding this comment

Uh oh!

liangyepianzhou left a comment

Choose a reason for hiding this comment

Uh oh!

BewareMyPower commented Nov 21, 2022

Uh oh!

Uh oh!

congbobo184 commented Nov 21, 2022

Uh oh!

BewareMyPower commented Nov 22, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

codecov-commenter commented Nov 16, 2022 •

edited

Loading

congbobo184 left a comment •

edited

Loading