[fix][broker]Fix chunked messages will be filtered by duplicating by liangyepianzhou · Pull Request #20948 · apache/pulsar

liangyepianzhou · 2023-08-07T15:40:28Z

Motivation

Make the chunk message function work properly when deduplication is enabled.

Modification

Only check and store the sequence ID of the last chunk in a chunk message.

For example:

    Chunk-1 sequence ID: 0, chunk ID: 0, total chunk: 2
    Chunk-2 sequence ID: 0, chunk ID: 1
    Chunk-3 sequence ID: 1, chunk ID: 0 total chunk: 3
    Chunk-4 sequence ID: 1, chunk ID: 1
    Chunk-5 sequence ID: 1, chunk ID: 1
    Chunk-6 sequence ID: 1, chunk ID: 2

Only store check and store the sequence ID of Chunk-2 and Chunk-6.
Add a property in the publishContext to determine whether this chunk is the last chunk when persistent completely.

publishContext.setProperty(IS_LAST_CHUNK, Boolean.FALSE);

Filter and ack duplicated chunks in a chunk message instead of discarding ctx.

For example:

    Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
    Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
    Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3
    Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
    Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5
    Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6

We should filter and ack chunk-4 and chunk-5.

Documentation

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

### Motivation Chunked messages use the same metadata, so all the chunked messages in a single message use the same sequence Id. And it will be recorded as duplicated messages. ``` private long updateMessageMetadataSequenceId(final MessageMetadata msgMetadata) { final long sequenceId; if (!msgMetadata.hasSequenceId()) { sequenceId = msgIdGenerator++; msgMetadata.setSequenceId(sequenceId); } else { sequenceId = msgMetadata.getSequenceId(); } return sequenceId; } ``` ### Modification Use different sequence id for chunk message.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java

heesung-sohn · 2023-08-07T18:11:07Z

@rdhabalia Please review this change. I think we have missed discussing the deduplication feature compatibility when introducing the chunking feature in the PIP(https://github.com/apache/pulsar/wiki/PIP-37:-Large-message-size-handling-in-Pulsar).

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

heesung-sohn · 2023-08-22T07:16:21Z

Does this pr require review? Have you raised a PIP for another solution?

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

BewareMyPower

Please create another test class like MessageChunkingDeduplicationTest for it for the following reasons:

MessageChunkingSharedTest was added to test chunking with Shared subscriptions. It's bad to add unrelated tests into this class.
You won't need to set conf.setBrokerDeduplicationEnabled(true) and restart the broker for each test method.

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingDeduplicationTest.java

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

…0948) Make the chunk message function work properly when deduplication is enabled. For example: ```markdown Chunk-1 sequence ID: 0, chunk ID: 0, total chunk: 2 Chunk-2 sequence ID: 0, chunk ID: 1 Chunk-3 sequence ID: 1, chunk ID: 0 total chunk: 3 Chunk-4 sequence ID: 1, chunk ID: 1 Chunk-5 sequence ID: 1, chunk ID: 1 Chunk-6 sequence ID: 1, chunk ID: 2 ``` Only store check and store the sequence ID of Chunk-2 and Chunk-6. **Add a property in the publishContext to determine whether this chunk is the last chunk when persistent completely.** ```java publishContext.setProperty(IS_LAST_CHUNK, Boolean.FALSE); ``` For example: ```markdown Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1 Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2 Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3 Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4 Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5 Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6 ``` We should filter and ack chunk-4 and chunk-5. (cherry picked from commit b0b13bc)

…0948) ## Motivation Make the chunk message function work properly when deduplication is enabled. ## Modification ### Only check and store the sequence ID of the last chunk in a chunk message. For example: ```markdown Chunk-1 sequence ID: 0, chunk ID: 0, total chunk: 2 Chunk-2 sequence ID: 0, chunk ID: 1 Chunk-3 sequence ID: 1, chunk ID: 0 total chunk: 3 Chunk-4 sequence ID: 1, chunk ID: 1 Chunk-5 sequence ID: 1, chunk ID: 1 Chunk-6 sequence ID: 1, chunk ID: 2 ``` Only store check and store the sequence ID of Chunk-2 and Chunk-6. **Add a property in the publishContext to determine whether this chunk is the last chunk when persistent completely.** ```java publishContext.setProperty(IS_LAST_CHUNK, Boolean.FALSE); ``` ### Filter and ack duplicated chunks in a chunk message instead of discarding ctx. For example: ```markdown Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1 Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2 Chunk-3 sequence ID: 0, chunk ID: 2, msgID: 1:3 Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4 Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5 Chunk-6 sequence ID: 0, chunk ID: 3, msgID: 1:6 ``` We should filter and ack chunk-4 and chunk-5.

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Aug 7, 2023

heesung-sohn reviewed Aug 7, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

poorbarcode requested a review from BewareMyPower August 9, 2023 02:24

optimize

30c9430

heesung-sohn reviewed Aug 10, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

liangyepianzhou added 2 commits August 10, 2023 20:18

optimize

02d7071

handle in broker side

d99ed6d

heesung-sohn reviewed Aug 11, 2023

View reviewed changes

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java Outdated Show resolved Hide resolved

heesung-sohn reviewed Aug 11, 2023

View reviewed changes

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java Outdated Show resolved Hide resolved

broker reload and message resend

be88b7b

heesung-sohn reviewed Aug 12, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

heesung-sohn reviewed Aug 12, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

Only check sequence ID

1c8eec3

heesung-sohn reviewed Aug 12, 2023

View reviewed changes

...r-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java Outdated Show resolved Hide resolved

filter duplicated message at consumer side

ab69604

codelipenghui added type/bug The PR fixed a bug or issue reported a bug category/functionality Some functions are not working such as get errors labels Aug 21, 2023

codelipenghui assigned liangyepianzhou Aug 21, 2023

codelipenghui added this to the 3.2.0 milestone Aug 21, 2023

codelipenghui added release/3.0.2 release/2.11.3 release/2.10.6 release/3.1.1 area/broker labels Aug 21, 2023

liangyepianzhou added 5 commits August 28, 2023 08:56

optimize uuid and checkstyle

1f8263d

check the last chunk ID

4f789ee

Merge remote-tracking branch 'apache/master' into deplicate_chunk

86c3f77

only store the sequence ID of the last chunk ID

2aaac45

Avoid flaky testing

8579c37

BewareMyPower reviewed Aug 30, 2023

View reviewed changes

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java Outdated Show resolved Hide resolved

Make comments indent aligned.

3e2267c

BewareMyPower requested changes Aug 30, 2023

View reviewed changes

BewareMyPower reviewed Aug 30, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java Outdated Show resolved Hide resolved

Add a new test class

7c813e9

BewareMyPower approved these changes Aug 30, 2023

View reviewed changes

poorbarcode approved these changes Aug 30, 2023

View reviewed changes

codelipenghui requested changes Aug 30, 2023

View reviewed changes

address some comments

233c5cc

codelipenghui approved these changes Aug 30, 2023

View reviewed changes

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java Show resolved Hide resolved

heesung-sohn reviewed Aug 30, 2023

View reviewed changes

heesung-sohn self-requested a review August 30, 2023 23:43

RobertIndie approved these changes Aug 31, 2023

View reviewed changes

heesung-sohn approved these changes Aug 31, 2023

View reviewed changes

liangyepianzhou merged commit b0b13bc into apache:master Aug 31, 2023

liangyepianzhou deleted the deplicate_chunk branch August 31, 2023 01:58

liangyepianzhou added the cherry-picked/branch-2.10 label Sep 4, 2023

Technoboy- added the cherry-picked/branch-3.1 label Sep 5, 2023

Technoboy- added the cherry-picked/branch-3.0 label Sep 5, 2023

BewareMyPower mentioned this pull request Oct 8, 2023

[Bug] Segmentation fault when sending messages after receiving an error apache/pulsar-client-cpp#325

Closed

2 tasks

shibd added the cherry-picked/branch-2.11 label Oct 22, 2023

Comments

Conversation

liangyepianzhou commented Aug 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modification

Only check and store the sequence ID of the last chunk in a chunk message.

Filter and ack duplicated chunks in a chunk message instead of discarding ctx.

Documentation

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

heesung-sohn commented Aug 7, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

heesung-sohn commented Aug 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

BewareMyPower left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

liangyepianzhou commented Aug 7, 2023 •

edited

Loading

heesung-sohn commented Aug 22, 2023 •

edited

Loading