[feature][broker] PIP 37: Support chunking with Shared subscription #16202

BewareMyPower · 2022-06-23T17:57:42Z

Motivation

https://github.com/apache/pulsar/wiki/PIP-37%3A-Large-message-size-handling-in-Pulsar#option-1-broker-caches-mapping-of-message-uuid-and-consumerid

The cons of option 1 described in the original proposal don't exist
for current code because broker keeps redelivered messages into
sorted map now.

Modifications

First of all, to avoid too many code changes, an EntryAndMetadata
class is introduced to bind the Entry with the associated
MessageMetadata to avoid parsing the metadata repeatedly. It also
implements the Entry interface, so this PR changes some
List<Entry> parameters to List<? extends Entry> so that a
List<EntryAndMetadata> argument can be accepted.

Then, a SharedConsumerAssignor is introduced to assign a list of
entries to all shared consumers.

Use a default selector to select the next consumer, like
PersistentDispatcherMultipleConsumers#getNextConsumer,
Each time a consumer is chosen, assign the entries in range
[i, i+permits) to the consumer except entries that have uuid:

If uuid is not cached, cache uuid -> consumer to indicate the
chunked message of this uuid must be dispatched to this consumer.
Otherwise, assign this entry to the owner consumer of the uuid.

The assign method returns a map that maps Consumer to
List<EntryAndMetadata>. The following logic is similar to the
Key_Shared dispatcher.

Finally, cancel the limit in ConsumerImpl.

Verifying this change

SharedConsumerAssignorTest is added to show how the assignor works
in detail.

MessageChunkingSharedTest is added to verify the Shared dispatcher
works on chunked messages, including:

Single producer sends chunked messages with various chunk count to a
consumer has a limited permits.
Single producer sends chunked messages to two consumers to verify
both they can receive chunked messages.
Produce interleaved chunks via PersistentTopic directly to simulate
multiple producers, and verify the new consumer can receive all
unacknowledged messages received by the old consumer.

Documentation

Check the box below or label this PR directly.

Need to update docs?

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

github-actions · 2022-06-23T17:58:13Z

@BewareMyPower Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

github-actions · 2022-06-23T17:59:30Z

@BewareMyPower Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.

RobertIndie

Overall looks good to me. Left some comments.

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

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

RobertIndie · 2022-06-24T12:36:09Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

-                long stickyKeyHash = getStickyKeyHash(entry);
-                addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
-                entry.release();
+            if (messagesForC < entryAndMetadataList.size()) {


When will messagesForC or consumer.getAvailablePermits() be less than entryAndMetadataList.size()?

I'm not sure, maybe the consumer's permits might change in the loop. It's added here for safety programming.

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

RobertIndie · 2022-06-30T03:56:11Z

This PR will fixes #7645

BewareMyPower · 2022-07-04T07:35:21Z

@codelipenghui @merlimat @rdhabalia @Jason918 @eolivelli @315157973 Could you take a look at this PR?

BewareMyPower · 2022-07-12T09:07:01Z

/pulsarbot rerun-failure-checks

BewareMyPower · 2022-07-15T09:43:04Z

@merlimat @lhotari @Jason918 @rdhabalia @hangc0276 @eolivelli @codelipenghui @massakam @congbobo184 @315157973 Could you take a second look at this PR?

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedConsumerAssignorTest.java

codelipenghui · 2022-07-29T02:10:46Z

@BewareMyPower Please help resolve the conflicts.

### Motivation https://github.com/apache/pulsar/wiki/PIP-37%3A-Large-message-size-handling-in-Pulsar#option-1-broker-caches-mapping-of-message-uuid-and-consumerid The cons of option 1 described in the original proposal don't exist for current code because broker keeps redelivered messages into **sorted** map now. ### Modifications First of all, to avoid too many code changes, an `EntryAndMetadata` class is introduced to bind the `Entry` with the associated `MessageMetadata` to avoid parsing the metadata repeatedly. It also implements the `Entry` interface, so this PR changes some `List<Entry>` parameters to `List<? extends Entry>` so that a `List<EntryAndMetadata>` argument can be accepted. Then, a `SharedConsumerAssignor` is introduced to assign a list of entries to all shared consumers. 1. Use a default selector to select the next consumer, like `PersistentDispatcherMultipleConsumers#getNextConsumer`, 2. Each time a consumer is chosen, assign the entries in range [i, i+permits) to the consumer except entries that have uuid: - If uuid is not cached, cache `uuid -> consumer` to indicate the chunked message of this uuid must be dispatched to this consumer. - Otherwise, assign this entry to the owner consumer of the uuid. The `assign` method returns a map that maps `Consumer` to `List<EntryAndMetadata>`. The following logic is similar to the Key_Shared dispatcher. Finally, cancel the limit in `ConsumerImpl`. ### Verifying this change `SharedConsumerAssignorTest` is added to show how the assignor works in detail. `MessageChunkingSharedTest` is added to verify the Shared dispatcher works on chunked messages, including: - Single producer sends chunked messages with various chunk count to a consumer has a limited permits. - Single producer sends chunked messages to two consumers to verify both they can receive chunked messages. - Produce interleaved chunks via `PersistentTopic` directly to simulate multiple producers, and verify the new consumer can receive all unacknowledged messages received by the old consumer. ### TODO We need to change the implementation of `ChunkMessageIdImpl` to make it possible for consumer to acknowledge all entries of a chunked message. Since this PR already includes many changes, I will do that later.

…pache#16202) * [feature][broker] PIP 37: Support chunking with Shared subscription ### Motivation https://github.com/apache/pulsar/wiki/PIP-37%3A-Large-message-size-handling-in-Pulsar#option-1-broker-caches-mapping-of-message-uuid-and-consumerid The cons of option 1 described in the original proposal don't exist for current code because broker keeps redelivered messages into **sorted** map now. ### Modifications First of all, to avoid too many code changes, an `EntryAndMetadata` class is introduced to bind the `Entry` with the associated `MessageMetadata` to avoid parsing the metadata repeatedly. It also implements the `Entry` interface, so this PR changes some `List<Entry>` parameters to `List<? extends Entry>` so that a `List<EntryAndMetadata>` argument can be accepted. Then, a `SharedConsumerAssignor` is introduced to assign a list of entries to all shared consumers. 1. Use a default selector to select the next consumer, like `PersistentDispatcherMultipleConsumers#getNextConsumer`, 2. Each time a consumer is chosen, assign the entries in range [i, i+permits) to the consumer except entries that have uuid: - If uuid is not cached, cache `uuid -> consumer` to indicate the chunked message of this uuid must be dispatched to this consumer. - Otherwise, assign this entry to the owner consumer of the uuid. The `assign` method returns a map that maps `Consumer` to `List<EntryAndMetadata>`. The following logic is similar to the Key_Shared dispatcher. Finally, cancel the limit in `ConsumerImpl`. ### Verifying this change `SharedConsumerAssignorTest` is added to show how the assignor works in detail. `MessageChunkingSharedTest` is added to verify the Shared dispatcher works on chunked messages, including: - Single producer sends chunked messages with various chunk count to a consumer has a limited permits. - Single producer sends chunked messages to two consumers to verify both they can receive chunked messages. - Produce interleaved chunks via `PersistentTopic` directly to simulate multiple producers, and verify the new consumer can receive all unacknowledged messages received by the old consumer.

BewareMyPower added type/feature The PR added a new feature or issue requested a new feature area/broker type/PIP labels Jun 23, 2022

BewareMyPower added this to the 2.11.0 milestone Jun 23, 2022

BewareMyPower requested review from merlimat, lhotari, Jason918, rdhabalia, hangc0276, eolivelli, codelipenghui, RobertIndie, massakam and congbobo184 June 23, 2022 17:57

BewareMyPower self-assigned this Jun 23, 2022

github-actions bot added the doc-label-missing label Jun 23, 2022

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch from a815c22 to b60557b Compare June 23, 2022 18:00

BewareMyPower added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Jun 23, 2022

RobertIndie reviewed Jun 24, 2022

View reviewed changes

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch from 8479c9c to 588c3c1 Compare June 28, 2022 06:00

Gleiphir2769 mentioned this pull request Jun 29, 2022

Support Large Message Size (PIP-37 / "chunking") apache/pulsar-client-go#456

Closed

RobertIndie approved these changes Jun 30, 2022

View reviewed changes

BewareMyPower requested a review from 315157973 July 4, 2022 07:34

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch from 588c3c1 to a89ef8a Compare July 11, 2022 08:46

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch 2 times, most recently from 89cc8ce to 788c422 Compare July 15, 2022 03:17

codelipenghui modified the milestones: 2.11.0, 2.12.0 Jul 26, 2022

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch from 788c422 to 43d7264 Compare July 28, 2022 09:17

codelipenghui approved these changes Jul 28, 2022

View reviewed changes

codelipenghui reviewed Jul 28, 2022

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedConsumerAssignorTest.java Show resolved Hide resolved

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch from 43d7264 to 76db14f Compare July 29, 2022 03:42

BewareMyPower added 4 commits August 1, 2022 10:24

Release the entry after it's added to the replay queue

916bd5d

Remove repeated code

f96d422

Add group to SharedConsumerAssignorTest

6fcc4cf

BewareMyPower force-pushed the bewaremypower/chunk-msg-for-shared branch from 76db14f to 6fcc4cf Compare August 1, 2022 02:24

Fix incorrect rebase

b98982a

codelipenghui merged commit b1a29b5 into apache:master Aug 2, 2022

labuladong mentioned this pull request Feb 21, 2023

[fix][doc] Shared subscription support chunking apache/pulsar-site#429

Merged

4 tasks

RobertIndie mentioned this pull request Mar 20, 2023

Large message size handling in Pulsar: Chunking for shared subscription #7645

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[feature][broker] PIP 37: Support chunking with Shared subscription #16202

[feature][broker] PIP 37: Support chunking with Shared subscription #16202

BewareMyPower commented Jun 23, 2022 •

edited

Loading

github-actions bot commented Jun 23, 2022

github-actions bot commented Jun 23, 2022

RobertIndie left a comment

RobertIndie Jun 24, 2022

BewareMyPower Jun 24, 2022

RobertIndie commented Jun 30, 2022

BewareMyPower commented Jul 4, 2022

BewareMyPower commented Jul 12, 2022

BewareMyPower commented Jul 15, 2022

codelipenghui commented Jul 29, 2022

[feature][broker] PIP 37: Support chunking with Shared subscription #16202

[feature][broker] PIP 37: Support chunking with Shared subscription #16202

Conversation

BewareMyPower commented Jun 23, 2022 • edited Loading

Motivation

Modifications

Verifying this change

Documentation

github-actions bot commented Jun 23, 2022

github-actions bot commented Jun 23, 2022

RobertIndie left a comment

Choose a reason for hiding this comment

RobertIndie Jun 24, 2022

Choose a reason for hiding this comment

BewareMyPower Jun 24, 2022

Choose a reason for hiding this comment

RobertIndie commented Jun 30, 2022

BewareMyPower commented Jul 4, 2022

BewareMyPower commented Jul 12, 2022

BewareMyPower commented Jul 15, 2022

codelipenghui commented Jul 29, 2022

BewareMyPower commented Jun 23, 2022 •

edited

Loading