limiting batch size to the minimum of the maxNumberOfMessages and maxSizeOfMessages #6865

avimas · 2020-05-03T12:29:26Z

Batch size is not limited to the minimum of the maxNumberOfMessages and maxSizeOfMessages from the BatchRecieve policy.

Batch size is not limited to the minimum of the maxNumberOfMessages and maxSizeOfMessages from the BatchRecieve policy.
I think this issue can be easly fixed by changing the canAdd function in MessagesImpl from:

protected boolean canAdd(Message<T> message) {
    if (this.maxNumberOfMessages <= 0 && this.maxSizeOfMessages <= 0L) {
        return true;
    } else {
        return this.maxNumberOfMessages > 0 && this.currentNumberOfMessages + 1 <= this.maxNumberOfMessages || this.maxSizeOfMessages > 0L &&   this.currentSizeOfMessages + (long)message.getData().length <= this.maxSizeOfMessages;
    }
}

to (changing the condintion in the else to && instead of ||):

protected boolean canAdd(Message<T> message) {
    if (this.maxNumberOfMessages <= 0 && this.maxSizeOfMessages <= 0L) {
        return true;
    } else {
        return (this.maxNumberOfMessages > 0 && this.currentNumberOfMessages + 1 <= this.maxNumberOfMessages) && (this.maxSizeOfMessages > 0L &&   this.currentSizeOfMessages + (long)message.getData().length <= this.maxSizeOfMessages);
    }
}

When the batch size is higher than the recieveQ of the consumer (I used a batch size of 3000 and a receiveQ of 500) I noticed the following issues:
a. In a mutliTopic (pattern) consumer the client stops receiving any messages I think it getting paused and never resumed when setting a timeout in the batch policy, only one batch is fetched and the client never resumed.

talked with @codelipenghui and advised to open this pull request

codelipenghui · 2020-05-06T14:21:46Z

/pulsarbot run-failure-checks

codelipenghui · 2020-05-07T06:42:36Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-11T07:30:46Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-11T10:37:19Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-11T14:21:41Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-12T00:32:52Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-12T04:16:46Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-12T06:13:55Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-12T08:30:07Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-12T13:45:00Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-13T00:47:15Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-13T02:54:11Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-13T05:07:20Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-13T05:42:59Z

would like to remove it from 2.5.2

jiazhai · 2020-05-13T11:06:04Z

/pulsarbot run-failure-checks

jiazhai · 2020-05-13T17:30:53Z

/pulsarbot run-failure-checks

codelipenghui · 2020-06-02T07:58:11Z

/pulsarbot run-failure-checks

codelipenghui · 2020-06-04T07:00:02Z

@avimas Could you please help rebase to the master branch? There are some CI test fix on master branch.

avimas · 2020-06-04T07:54:30Z

@avimas Could you please help rebase to the master branch? There are some CI test fix on master branch.

Done!

codelipenghui · 2020-06-05T00:03:11Z

@avimas Please rebase again, I can't push to your master branch. It's better to checkout -b a new branch to create a pull request. #7173 has fix some issues with CI tests.

avimas · 2020-06-05T19:15:03Z

@codelipenghui rebase done, I'll make sure to use a new branch next time.

codelipenghui · 2020-06-06T23:26:06Z

/pulsarbot run-failure-checks

codelipenghui · 2020-06-08T04:10:40Z

@avimas Could you please take a look at the failed CI tests? look related to this PR.

codelipenghui · 2020-06-08T06:53:10Z

move to 2.7.0 first.

…SizeOfMessages from the BatchReceive policy

…calling internalBatchReceiveAsync

sijie · 2020-06-10T05:37:56Z

/pulsarbot run-failure-checks

codelipenghui · 2020-06-10T06:07:19Z

/pulsarbot run-failure-checks

@codelipenghui

…SizeOfMessages (#6865) Batch size is not limited to the minimum of the maxNumberOfMessages and maxSizeOfMessages from the BatchRecieve policy. 1. Batch size is not limited to the minimum of the maxNumberOfMessages and maxSizeOfMessages from the BatchRecieve policy. I think this issue can be easly fixed by changing the canAdd function in MessagesImpl from: ``` protected boolean canAdd(Message<T> message) { if (this.maxNumberOfMessages <= 0 && this.maxSizeOfMessages <= 0L) { return true; } else { return this.maxNumberOfMessages > 0 && this.currentNumberOfMessages + 1 <= this.maxNumberOfMessages || this.maxSizeOfMessages > 0L && this.currentSizeOfMessages + (long)message.getData().length <= this.maxSizeOfMessages; } } ``` to (changing the condintion in the else to && instead of ||): ``` protected boolean canAdd(Message<T> message) { if (this.maxNumberOfMessages <= 0 && this.maxSizeOfMessages <= 0L) { return true; } else { return (this.maxNumberOfMessages > 0 && this.currentNumberOfMessages + 1 <= this.maxNumberOfMessages) && (this.maxSizeOfMessages > 0L && this.currentSizeOfMessages + (long)message.getData().length <= this.maxSizeOfMessages); } } ``` 2. When the batch size is higher than the recieveQ of the consumer (I used a batch size of 3000 and a receiveQ of 500) I noticed the following issues: a. In a mutliTopic (pattern) consumer the client stops receiving any messages I think it getting paused and never resumed when setting a timeout in the batch policy, only one batch is fetched and the client never resumed. talked with @codelipenghui and advised to open this pull request (cherry picked from commit 941ddd6)

@codelipenghui

…SizeOfMessages (apache#6865) Batch size is not limited to the minimum of the maxNumberOfMessages and maxSizeOfMessages from the BatchRecieve policy. 1. Batch size is not limited to the minimum of the maxNumberOfMessages and maxSizeOfMessages from the BatchRecieve policy. I think this issue can be easly fixed by changing the canAdd function in MessagesImpl from: ``` protected boolean canAdd(Message<T> message) { if (this.maxNumberOfMessages <= 0 && this.maxSizeOfMessages <= 0L) { return true; } else { return this.maxNumberOfMessages > 0 && this.currentNumberOfMessages + 1 <= this.maxNumberOfMessages || this.maxSizeOfMessages > 0L && this.currentSizeOfMessages + (long)message.getData().length <= this.maxSizeOfMessages; } } ``` to (changing the condintion in the else to && instead of ||): ``` protected boolean canAdd(Message<T> message) { if (this.maxNumberOfMessages <= 0 && this.maxSizeOfMessages <= 0L) { return true; } else { return (this.maxNumberOfMessages > 0 && this.currentNumberOfMessages + 1 <= this.maxNumberOfMessages) && (this.maxSizeOfMessages > 0L && this.currentSizeOfMessages + (long)message.getData().length <= this.maxSizeOfMessages); } } ``` 2. When the batch size is higher than the recieveQ of the consumer (I used a batch size of 3000 and a receiveQ of 500) I noticed the following issues: a. In a mutliTopic (pattern) consumer the client stops receiving any messages I think it getting paused and never resumed when setting a timeout in the batch policy, only one batch is fetched and the client never resumed. talked with @codelipenghui and advised to open this pull request

codelipenghui approved these changes May 3, 2020

View reviewed changes

codelipenghui requested review from jiazhai and sijie May 3, 2020 12:57

codelipenghui assigned avimas May 3, 2020

codelipenghui added the type/bug The PR fixed a bug or issue reported a bug label May 3, 2020

codelipenghui added this to the 2.6.0 milestone May 3, 2020

codelipenghui added the release/2.5.2 label May 3, 2020

sijie approved these changes May 6, 2020

View reviewed changes

jiazhai closed this May 12, 2020

jiazhai reopened this May 12, 2020

jiazhai approved these changes May 13, 2020

View reviewed changes

jiazhai removed the release/2.5.2 label May 13, 2020

codelipenghui changed the title ~~limiting batch size to the minimum of the maxNumberOfMessages and max…~~ limiting batch size to the minimum of the maxNumberOfMessages and maxSizeOfMessages Jun 4, 2020

avimas force-pushed the master branch from 268874d to d904843 Compare June 4, 2020 07:53

avimas force-pushed the master branch from d904843 to 2eaab90 Compare June 5, 2020 19:03

codelipenghui modified the milestones: 2.6.0, 2.7.0 Jun 8, 2020

avimas force-pushed the master branch from 2eaab90 to 7768887 Compare June 9, 2020 15:48

avim added 3 commits June 9, 2020 21:06

limiting batch size to the minimum of the maxNumberOfMessages and max…

759c42f

…SizeOfMessages from the BatchReceive policy

Resuming paused consumers if needed In MultiTopicsConsumerImpl after …

9051456

…calling internalBatchReceiveAsync

fix failed test and add unitest to verify fix

9ea5240

avimas force-pushed the master branch from 7768887 to 9ea5240 Compare June 9, 2020 18:20

codelipenghui approved these changes Jun 10, 2020

View reviewed changes

codelipenghui added the release/2.6.1 label Jun 10, 2020

codelipenghui merged commit 941ddd6 into apache:master Jun 10, 2020

e-marchand-exensa mentioned this pull request Aug 27, 2020

[Java] [v2.5.2] BatchReceivePolicy not strictly complied by ConsumerBase and MessageImpl #7696

Closed

limiting batch size to the minimum of the maxNumberOfMessages and maxSizeOfMessages #6865

limiting batch size to the minimum of the maxNumberOfMessages and maxSizeOfMessages #6865

Uh oh!

Conversation

avimas commented May 3, 2020 • edited by jiazhai Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codelipenghui commented May 6, 2020

Uh oh!

codelipenghui commented May 7, 2020

Uh oh!

jiazhai commented May 11, 2020

Uh oh!

jiazhai commented May 11, 2020

Uh oh!

jiazhai commented May 11, 2020

Uh oh!

jiazhai commented May 12, 2020

Uh oh!

jiazhai commented May 12, 2020

Uh oh!

jiazhai commented May 12, 2020

Uh oh!

jiazhai commented May 12, 2020

Uh oh!

jiazhai commented May 12, 2020

Uh oh!

jiazhai commented May 13, 2020

Uh oh!

jiazhai commented May 13, 2020

Uh oh!

jiazhai commented May 13, 2020

Uh oh!

jiazhai commented May 13, 2020

Uh oh!

jiazhai commented May 13, 2020

Uh oh!

jiazhai commented May 13, 2020

Uh oh!

codelipenghui commented Jun 2, 2020

Uh oh!

codelipenghui commented Jun 4, 2020

Uh oh!

avimas commented Jun 4, 2020

Uh oh!

codelipenghui commented Jun 5, 2020

Uh oh!

avimas commented Jun 5, 2020

Uh oh!

codelipenghui commented Jun 6, 2020

Uh oh!

codelipenghui commented Jun 8, 2020

Uh oh!

codelipenghui commented Jun 8, 2020

Uh oh!

sijie commented Jun 10, 2020

Uh oh!

codelipenghui commented Jun 10, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

avimas commented May 3, 2020 •

edited by jiazhai

Loading