[pulsar-broker] Dispatch batch messages according consumer permits #7266

rdhabalia · 2020-06-13T01:45:46Z

Motivation

Right now, dispatching of batch messages is not predictable for a shared-subscription. Dispatcher doesn't consider number of batch messages in an entry which can cause higher number of message delivery than consumer receiver-queue.
For example:
If a topic has message-entries with 100 batched message in an entry.
If consumer has receiver queue=40 and has available-permits=20. then broker dispatches 20 entries with total 2000 (20*100) messages in it. because of that consumer's will high number of messages than expected and we can see available-permits in negative.

Modification

Find out average number of batch messages in an entry
Dispatch messages considering total batch-messages in an entry

Result

It will avoid dispatching higher number of messages than consumer's available-permit and will fix -ve available-permits.

codelipenghui · 2020-06-13T12:39:48Z

@rdhabalia Looks like the same purpose as #6719

rdhabalia · 2020-06-13T18:29:43Z

@codelipenghui
I saw #6719 but it doesn't solve fundamental problem of dispatching. #6719 tries to predict how many messages should be read by considering total-available permits and total batch messages. but it's still not controlling while dispatching messages.
for example: there are 10 consumers connected with 100 permits each and one batch message contains 100 messages. Now, broker reads 10 messages based on both factors but broker doesn't know number of batch-messages in each message so, it will dispatch all 10 messages to the consumer which has 100 permits which is the actual issue because broker dispatched 10*100=1K message to the consumer.

codelipenghui · 2020-06-16T01:16:13Z

@rdhabalia Thanks, I now know this problem that you want to fix.

codelipenghui · 2020-06-16T01:19:18Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java

+    /**
+     * Get number of batch messages into the entry.
+     * @return
+     */
+    int getNumberOfBatchMessages();
+
+    /**
+     * Set number of batch messages into the entry.
+     * @param numberOfBatchMessages
+     */
+    void setNumberOfBatchMessages(int numberOfBatchMessages);
+
+    /**
+     * Set data size of the entry.
+     * @param totalSizeInBytes
+     */
+    void setTotalSizeInBytes(int totalSizeInBytes);
+
+    /**
+     * Set total number of chunked messages in entry.
+     * @param totalChunkedMessages
+     */
+    void setTotalChunkedMessages(int totalChunkedMessages);
+
+    /**
+     * Get total data size of the entry.
+     * @return
+     */
+    int getTotalSizeInBytes();
+
+    /**
+     * Get total number of chuncked messages.
+     * @return
+     */
+    int getTotalChunkedMessages();


I think we better not add message related concepts to the entry. Maybe we can add a context carrier in the entry. What do you think?

Hi @codelipenghui , could you please provide a little more detail on what you mean by a "context carrier"?

I agree with @codelipenghui
it is unfortunate to add these methods here.
we are going to use them (and the underlying fields) only on the Dispatcher.

it would be better (for instance) to have some EntryWithMetatadata object in Dispatcher that wraps the Entry and adds these fields

Huanli-Meng · 2020-06-19T00:11:21Z

I. think it should open a doc issue to update docs for this PR. @jennifer88huang , could you help open an doc issue for this PR?

jiazhai · 2020-06-28T01:08:11Z

Thanks @rdhabalia. overall lgtm. Would you please help on @codelipenghui 's last comments?

Jennifer88huang-zz · 2020-06-28T01:58:29Z

I. think it should open a doc issue to update docs for this PR. @jennifer88huang , could you help open an doc issue for this PR?

@Huanli-Meng Thanks for informing me on this. We can add a label "doc-required" in the Labels list, I've created the issue.

Anonymitaet · 2021-03-09T04:29:27Z

@rdhabalia thanks for your work.

Any progress for this PR?
Could you please help add docs accordingly? Thanks

Anonymitaet · 2021-04-09T07:37:04Z

@rdhabalia thanks for your work.

Any progress for this PR?
Could you please help add docs accordingly? Thanks

rdhabalia · 2021-04-28T03:24:48Z

I will rebase this soon.

Anonymitaet · 2021-04-28T03:34:27Z

@rdhabalia could you please help add docs accordingly? Then I would like to help review, thanks.

eolivelli

LGTM

eolivelli · 2021-05-15T05:26:42Z

@merlimat do you want to take a look as well?

lhotari

LGTM

lhotari · 2021-05-17T08:36:04Z

Closing and reopening this PR to get the fix for a flaky test.

…pache#7266) Right now, dispatching of batch messages is not predictable for a shared-subscription. Dispatcher doesn't consider number of batch messages in an entry which can cause higher number of message delivery than consumer receiver-queue. For example: If a topic has message-entries with 100 batched message in an entry. If consumer has receiver queue=40 and has available-permits=20. then broker dispatches 20 entries with total 2000 (20*100) messages in it. because of that consumer's will high number of messages than expected and we can see available-permits in negative. - Find out average number of batch messages in an entry - Dispatch messages considering total batch-messages in an entry It will avoid dispatching higher number of messages than consumer's available-permit and will fix -ve available-permits. (cherry picked from commit a6aed55)

…rmits (apache#7266)" This reverts commit 5d7ef69.

devinbost · 2021-05-17T19:35:54Z

@rdhabalia Nice work!

…pache#7266) Right now, dispatching of batch messages is not predictable for a shared-subscription. Dispatcher doesn't consider number of batch messages in an entry which can cause higher number of message delivery than consumer receiver-queue. For example: If a topic has message-entries with 100 batched message in an entry. If consumer has receiver queue=40 and has available-permits=20. then broker dispatches 20 entries with total 2000 (20*100) messages in it. because of that consumer's will high number of messages than expected and we can see available-permits in negative. - Find out average number of batch messages in an entry - Dispatch messages considering total batch-messages in an entry It will avoid dispatching higher number of messages than consumer's available-permit and will fix -ve available-permits. (cherry picked from commit a6aed55)

Fixes #10813 The issue is introduced by #7266, it only affects the master branch. ### Motivation 1. Add wrapperOffset to make sure get the correct batch size from the metadata 2. Fix the issue that using (batch count / avgBatchSizePerMsg) to calculate messages for the consumer it should be (messages / avgBatchSizePerMsg) ### Verifying this change * The test case is to simulate dispatch batches with different batch size to the consumer. * 1. The consumer has 1000 available permits * 2. The producer send batches with different batch size * * According the batch average size dispatching, the broker will dispatch all the batches to the consumer

Fixes apache#10813 The issue is introduced by apache#7266, it only affects the master branch. ### Motivation 1. Add wrapperOffset to make sure get the correct batch size from the metadata 2. Fix the issue that using (batch count / avgBatchSizePerMsg) to calculate messages for the consumer it should be (messages / avgBatchSizePerMsg) ### Verifying this change * The test case is to simulate dispatch batches with different batch size to the consumer. * 1. The consumer has 1000 available permits * 2. The producer send batches with different batch size * * According the batch average size dispatching, the broker will dispatch all the batches to the consumer (cherry picked from commit 4f23767)

…pache#7266) ### Motivation Right now, dispatching of batch messages is not predictable for a shared-subscription. Dispatcher doesn't consider number of batch messages in an entry which can cause higher number of message delivery than consumer receiver-queue. For example: If a topic has message-entries with 100 batched message in an entry. If consumer has receiver queue=40 and has available-permits=20. then broker dispatches 20 entries with total 2000 (20*100) messages in it. because of that consumer's will high number of messages than expected and we can see available-permits in negative. ### Modification - Find out average number of batch messages in an entry - Dispatch messages considering total batch-messages in an entry ### Result It will avoid dispatching higher number of messages than consumer's available-permit and will fix -ve available-permits.

Fixes apache#10813 The issue is introduced by apache#7266, it only affects the master branch. ### Motivation 1. Add wrapperOffset to make sure get the correct batch size from the metadata 2. Fix the issue that using (batch count / avgBatchSizePerMsg) to calculate messages for the consumer it should be (messages / avgBatchSizePerMsg) ### Verifying this change * The test case is to simulate dispatch batches with different batch size to the consumer. * 1. The consumer has 1000 available permits * 2. The producer send batches with different batch size * * According the batch average size dispatching, the broker will dispatch all the batches to the consumer

…pache#7266) ### Motivation Right now, dispatching of batch messages is not predictable for a shared-subscription. Dispatcher doesn't consider number of batch messages in an entry which can cause higher number of message delivery than consumer receiver-queue. For example: If a topic has message-entries with 100 batched message in an entry. If consumer has receiver queue=40 and has available-permits=20. then broker dispatches 20 entries with total 2000 (20*100) messages in it. because of that consumer's will high number of messages than expected and we can see available-permits in negative. ### Modification - Find out average number of batch messages in an entry - Dispatch messages considering total batch-messages in an entry ### Result It will avoid dispatching higher number of messages than consumer's available-permit and will fix -ve available-permits.

Fixes apache#10813 The issue is introduced by apache#7266, it only affects the master branch. ### Motivation 1. Add wrapperOffset to make sure get the correct batch size from the metadata 2. Fix the issue that using (batch count / avgBatchSizePerMsg) to calculate messages for the consumer it should be (messages / avgBatchSizePerMsg) ### Verifying this change * The test case is to simulate dispatch batches with different batch size to the consumer. * 1. The consumer has 1000 available permits * 2. The producer send batches with different batch size * * According the batch average size dispatching, the broker will dispatch all the batches to the consumer

… instead ### Motivation apache#7266 introduced the `EntryWrapper` to store the `Entry` object and the associated `MessageMetadata` if it exists. However, the `getEntry` field of `EntryWrapper` is never used. There is no need to allocate memory for `EntryWrapper`, even if it's allocated from the recycler pool. ### Modifications - Calculate the remaining messages without creating `EntryWrapper` object, just iterate over the parsed message metadata list. - Pass an optional `MessageMetadata` array to `filterEntriesForConsumer` and add the JavaDocs for these two parameters. After that, Remove unused `EntryWrapper` and `updateEntryWrapperWithMetadata`. This PR uses functional programming style to make code more simple and clear. ### Verifying this change - [ ] Make sure that the change passes the CI checks. This change is already covered by existing tests, such as `testBatchMessageDispatchingAccordingToPermits`. ### 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) - [x] `doc-not-needed` (Please explain why) - [ ] `doc` (Your PR contains doc changes) - [ ] `doc-complete` (Docs have been already added)

… instead (#15967) ### Motivation #7266 introduced the `EntryWrapper` to store the `Entry` object and the associated `MessageMetadata` if it exists. However, the `getEntry` field of `EntryWrapper` is never used. There is no need to allocate memory for `EntryWrapper`, even if it's allocated from the recycler pool. ### Modifications - Calculate the remaining messages without creating `EntryWrapper` object, just iterate over the parsed message metadata list. - Pass an optional `MessageMetadata` array to `filterEntriesForConsumer` and add the JavaDocs for these two parameters. After that, Remove unused `EntryWrapper` and `updateEntryWrapperWithMetadata`. This PR uses functional programming style to make code more simple and clear.

… instead (apache#15967) apache#7266 introduced the `EntryWrapper` to store the `Entry` object and the associated `MessageMetadata` if it exists. However, the `getEntry` field of `EntryWrapper` is never used. There is no need to allocate memory for `EntryWrapper`, even if it's allocated from the recycler pool. - Calculate the remaining messages without creating `EntryWrapper` object, just iterate over the parsed message metadata list. - Pass an optional `MessageMetadata` array to `filterEntriesForConsumer` and add the JavaDocs for these two parameters. After that, Remove unused `EntryWrapper` and `updateEntryWrapperWithMetadata`. This PR uses functional programming style to make code more simple and clear. (cherry picked from commit c33b12d)

rdhabalia added the area/broker label Jun 13, 2020

rdhabalia added this to the 2.7.0 milestone Jun 13, 2020

rdhabalia requested review from merlimat, maskit, sijie, jiazhai, aahmed-se, jai1, codelipenghui, nkurihar and massakam June 13, 2020 01:45

rdhabalia self-assigned this Jun 13, 2020

codelipenghui reviewed Jun 16, 2020

View reviewed changes

Jennifer88huang-zz added the doc-required Your PR changes impact docs and you will update later. label Jun 28, 2020

Jennifer88huang-zz mentioned this pull request Jun 28, 2020

[docs] Add docs for "Dispatch batch messages according consumer permits" #7378

Closed

sijie mentioned this pull request Jun 28, 2020

ISSUE-7378: [docs] Add docs for "Dispatch batch messages according consumer permits" streamnative/pulsar-archived#1148

Closed

codelipenghui modified the milestones: 2.7.0, 2.8.0 Nov 4, 2020

devinbost mentioned this pull request Mar 18, 2021

Catastrophic frequent random subscription freezes, especially on high-traffic topics. #6054

Closed

lhotari mentioned this pull request Apr 28, 2021

[Broker] Make Persistent*DispatcherMultipleConsumers.readMoreEntries synchronized #10413

Merged

rdhabalia force-pushed the batch_dispatch1 branch from 20fbcc9 to 3cb0334 Compare May 15, 2021 01:20

Remove entry api and added entry-wrapper

739c32f

rdhabalia force-pushed the batch_dispatch1 branch from 3cb0334 to 739c32f Compare May 15, 2021 01:23

eolivelli approved these changes May 15, 2021

View reviewed changes

eolivelli requested a review from codelipenghui May 15, 2021 05:26

lhotari approved these changes May 17, 2021

View reviewed changes

lhotari closed this May 17, 2021

lhotari reopened this May 17, 2021

codelipenghui approved these changes May 17, 2021

View reviewed changes

codelipenghui merged commit a6aed55 into apache:master May 17, 2021

eolivelli added a commit to datastax/pulsar that referenced this pull request May 17, 2021

Revert "[pulsar-broker] Dispatch batch messages according consumer pe…

d001672

…rmits (apache#7266)" This reverts commit 5d7ef69.

codelipenghui mentioned this pull request Jun 4, 2021

Fix consumer stuck issue due to reuse entry wrapper. #10824

Merged

BewareMyPower mentioned this pull request Jun 8, 2022

[refactor][broker] Remove EntryWrapper usages and use MessageMetadata instead #15967

Merged

5 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[pulsar-broker] Dispatch batch messages according consumer permits #7266

[pulsar-broker] Dispatch batch messages according consumer permits #7266

rdhabalia commented Jun 13, 2020

codelipenghui commented Jun 13, 2020

rdhabalia commented Jun 13, 2020

codelipenghui commented Jun 16, 2020

codelipenghui Jun 16, 2020

devinbost Apr 28, 2021

eolivelli May 13, 2021

Huanli-Meng commented Jun 19, 2020

jiazhai commented Jun 28, 2020

Jennifer88huang-zz commented Jun 28, 2020 •

edited

Loading

Anonymitaet commented Mar 9, 2021

Anonymitaet commented Apr 9, 2021

rdhabalia commented Apr 28, 2021

Anonymitaet commented Apr 28, 2021

eolivelli left a comment

eolivelli commented May 15, 2021

lhotari left a comment

lhotari commented May 17, 2021 •

edited

Loading

devinbost commented May 17, 2021

[pulsar-broker] Dispatch batch messages according consumer permits #7266

[pulsar-broker] Dispatch batch messages according consumer permits #7266

Conversation

rdhabalia commented Jun 13, 2020

Motivation

Modification

Result

codelipenghui commented Jun 13, 2020

rdhabalia commented Jun 13, 2020

codelipenghui commented Jun 16, 2020

codelipenghui Jun 16, 2020

Choose a reason for hiding this comment

devinbost Apr 28, 2021

Choose a reason for hiding this comment

eolivelli May 13, 2021

Choose a reason for hiding this comment

Huanli-Meng commented Jun 19, 2020

jiazhai commented Jun 28, 2020

Jennifer88huang-zz commented Jun 28, 2020 • edited Loading

Anonymitaet commented Mar 9, 2021

Anonymitaet commented Apr 9, 2021

rdhabalia commented Apr 28, 2021

Anonymitaet commented Apr 28, 2021

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli commented May 15, 2021

lhotari left a comment

Choose a reason for hiding this comment

lhotari commented May 17, 2021 • edited Loading

devinbost commented May 17, 2021

Jennifer88huang-zz commented Jun 28, 2020 •

edited

Loading

lhotari commented May 17, 2021 •

edited

Loading