[ISSUE #6205] Support Async pull for DefaultLitePullConsumer #6417

buptxxb65 · 2023-03-20T18:10:33Z

Make sure set the target branch to develop

What is the purpose of the change

fixes #6205
Support the new feature mentioned in it.

fix #

Brief changelog

Support the Async pull function by using PullCallback.

Currently, it has not been actually invoked. Since I'm confused when to use it(When the number of partitions/queues assigned to the client is greater than a specific number?).

I'll further modified it in the future. Looking forward to your suggestions : )

Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

codecov-commenter · 2023-03-21T00:31:33Z

Codecov Report

Merging #6417 (44dd532) into develop (f1e02e8) will increase coverage by 0.09%.
The diff coverage is 21.62%.

@@              Coverage Diff              @@
##             develop    #6417      +/-   ##
=============================================
+ Coverage      43.05%   43.14%   +0.09%     
- Complexity      8911     9014     +103     
=============================================
  Files           1105     1107       +2     
  Lines          77970    78321     +351     
  Branches       10146    10211      +65     
=============================================
+ Hits           33570    33794     +224     
- Misses         40190    40290     +100     
- Partials        4210     4237      +27

Impacted Files	Coverage Δ
...ent/impl/consumer/DefaultLitePullConsumerImpl.java	`64.33% <21.62%> (-3.62%)`	⬇️

... and 73 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

RongtongJin · 2023-03-21T00:41:43Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

+        public PullTaskImpl(final MessageQueue messageQueue, CommunicationMode communicationMode) {
+            this.messageQueue = messageQueue;
+            this.communicationMode = communicationMode;


Good job! How do users set whether the pullTask is asynchronous or synchronous?

Thanks for your comment! I have the same confusion about the entrance of asynchronous pullTask.

As you see, I create another constructor here. When user creates the PullTaskImpl with "Async", the pullTask will run in the asynchronous way. However, I find that the PullTaskImpl only be created in other methods (e.g. seek(), startPullTask()) in this class.
If we let user decide which way to pull the tasks, we need to add a new parameter "communicationMode" for all upstream functions or in the class DefaultLitePullConsumerImpl (maybe better? just a private parameter with a set() function). I think either of them will make the code more complex.

This is my thought on the entrance design, but as I'm a freshman in RocketMQ, there may be some areas where I have not considered thoroughly. I would appreciate your advice and guidance on this.

You are right. It is not appropriate for users to decide whether to synchronize or asynchronously. IMO, we should replace the current synchronous pull implementation after the asynchronous pull is stable.

Thank you for your approval! Based on your suggestion, I think we can keep this constructor to facilitate our testing of asynchronous pull. Besides, what else can I do for this PR?

Thank you for your approval! Based on your suggestion, I think we can keep this constructor to facilitate our testing of asynchronous pull. Besides, what else can I do for this PR?

Go ahead, you can fix the problem of the comments, then we should do some tests. :-)

I have fixed the problem mentioned in comments. Thanks for your effort! Please let me know if there is anything I can do for it.

RongtongJin · 2023-03-22T00:39:38Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

-                    updatePullOffset(messageQueue, pullResult.getNextBeginOffset(), processQueue);
                } catch (InterruptedException interruptedException) {
                    log.warn("Polling thread was interrupted.", interruptedException);
                } catch (Throwable e) {


If CommunicationMode is ASYNC, these code shoud be performed in callback.

THX! I'll correct it

RongtongJin · 2023-03-22T00:42:55Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

+        public PullTaskImpl(final MessageQueue messageQueue, CommunicationMode communicationMode) {
+            this.messageQueue = messageQueue;
+            this.communicationMode = communicationMode;


You are right. It is not appropriate for users to decide whether to synchronize or asynchronously. IMO, we should replace the current synchronous pull implementation after the asynchronous pull is stable.

guyinyou · 2023-03-22T08:45:22Z

If it is changed to asynchronous, this part of the code also needs to be moved to the callback

buptxxb65 · 2023-03-23T02:40:28Z

If it is changed to asynchronous, this part of the code also needs to be moved to the callback

THX! I'll correct it

github-advanced-security · 2023-03-24T02:27:01Z

You have successfully added a new CodeQL configuration .github/workflows/codeql_analysis.yml:CodeQL-Build. As part of the setup process, we have scanned this repository and found 9 existing alerts. Please check the repository Security tab to see all alerts.

guyinyou · 2023-03-24T03:44:18Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

+                if (e instanceof MQBrokerException && ((MQBrokerException) e).getResponseCode() == ResponseCode.FLOW_CONTROL) {
+                    pullDelayTimeMills = PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL;
+                } else {
+                    pullDelayTimeMills = pullTimeDelayMillsWhenException;


The time for the next scheduling is changed here, but there is no scheduling

Thank you for pointing it out. I will correct it as soon as possible.

Thank you for pointing it out. I will correct it as soon as possible.

Go ahead! :-)

miles-ton · 2023-03-24T17:40:04Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

+        this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, true, timeout, pullCallback);
+    }
+
+    private void pullAsyncImpl(MessageQueue mq, SubscriptionData subscriptionData, long offset, int maxNums,


is it possible to extract the same code as existed method pullSyncImpl()

Thanks for your comment! In fact, there are 11 lines of duplicated code in pullSyncImpl() and pullAsyncImpl(). So we can extract a method for them. However, I notice that in DefaultMQPullConsumer, these two functions are written in the same way(duplicated code).

"I retain a little. I think retain some flavor of RocketMQ can let you know you are reading RocketMQ's code." 🐶

RongtongJin · 2023-03-31T08:23:07Z

Hi @buptxxb65, could you do some tests and post a report？If you need any help, plz feel free to contact me.

buptxxb65 · 2023-04-02T06:11:25Z

Hi @buptxxb65, could you do some tests and post a report？If you need any help, plz feel free to contact me.

Sorry for late reply. Sure, I will do some tests for it. However, there is an examination on Tuesday(4.4), so I need to work on the tests and reports after it. I hope it won't be too late.

RongtongJin · 2023-04-03T09:39:51Z

Hi @buptxxb65, could you do some tests and post a report？If you need any help, plz feel free to contact me.

Sorry for late reply. Sure, I will do some tests for it. However, there is an examination on Tuesday(4.4), so I need to work on the tests and reports after it. I hope it won't be too late.

Of course it doesn't matter. Look forward to your test.

buptxxb65 · 2023-04-07T10:05:55Z

Hi @RongtongJin , I created an example for the DefaultLitePullConsumer. To use this example testing the Async pull, we need to change the line 680 in DefaultLitePullConsumerImpl into PullTaskImpl pullTask = new PullTaskImpl(messageQueue, CommunicationMode.ASYNC);

Here is the conclusion of its performance.

When the message is consumed successfully,
Different queues will run in concurrently, while the message in the same queue will be consumed orderly. We can check the QueueId and QueueOffset of messageExts to confirm it. I think this meets our expectations.
When the message is consumed unsuccessfully,
The exception will call the onException() in the pullCallback we defined in line 989. The pullDelayTimeMills will be changed according to the type of the exception. It will be set as the delay of the scheduledThreadPoolExecutor.

That's all of the test. I would appreciate if you have any better idea of the test and I'm happy to do anything I can to improve it.

RongtongJin · 2023-03-31T08:20:36Z

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

+            }
+        }
+
+        private void executeThreadWithDelay() {


How about rename to schedulePullTaskWithDelay？

RongtongJin · 2023-04-10T09:39:40Z

example/src/main/java/org/apache/rocketmq/example/simple/LitePullConsumerAsync.java

@@ -0,0 +1,42 @@
+package org.apache.rocketmq.example.simple;


IMO, no need for a new example.

Yes, I totally agree with you so I'll remove this example tonight. And I want to make sure my understanding is correct.

Since we haven't figure out a suitable way to start an Async pull, we can only change the constructor method of PullTaskImpl in seek(). Therefore, in users' view (example view), they just call the seek() function. Maybe we can add a new example after we separate the Async pull and Sync pull.

RongtongJin · 2023-04-10T09:50:26Z

Hi @RongtongJin , I created an example for the DefaultLitePullConsumer. To use this example testing the Async pull, we need to change the line 680 in DefaultLitePullConsumerImpl into PullTaskImpl pullTask = new PullTaskImpl(messageQueue, CommunicationMode.ASYNC);

Here is the conclusion of its performance.

When the message is consumed successfully,
Different queues will run in concurrently, while the message in the same queue will be consumed orderly. We can check the QueueId and QueueOffset of messageExts to confirm it. I think this meets our expectations.

When the message is consumed unsuccessfully,
The exception will call the onException() in the pullCallback we defined in line 989. The pullDelayTimeMills will be changed according to the type of the exception. It will be set as the delay of the scheduledThreadPoolExecutor.

That's all of the test. I would appreciate if you have any better idea of the test and I'm happy to do anything I can to improve it.

Firstly, testing needs to verify whether it can solve the problem described in the issue.

buptxxb65 · 2023-04-12T06:59:54Z

Firstly, testing needs to verify whether it can solve the problem described in the issue.

In achievement aspect, I think the testing has verified that it can successfully asynchronous pull messages. Do you mean I need to verify whether the Async pull will behave better than Sync pull under the special situation? Like the number of partitions/queues assigned to the client is much more greater than 20.

guyinyou · 2023-04-12T07:54:30Z

Firstly, testing needs to verify whether it can solve the problem described in the issue.

In achievement aspect, I think the testing has verified that it can successfully asynchronous pull messages. Do you mean I need to verify whether the Async pull will behave better than Sync pull under the special situation? Like the number of partitions/queues assigned to the client is much more greater than 20.

I think @RongtongJin means to verify the performance in some abnormal scenarios, such as restarting the broker during the process of pulling messages to see if it works normally

buptxxb65 · 2023-04-12T17:28:41Z

Firstly, testing needs to verify whether it can solve the problem described in the issue.

In achievement aspect, I think the testing has verified that it can successfully asynchronous pull messages. Do you mean I need to verify whether the Async pull will behave better than Sync pull under the special situation? Like the number of partitions/queues assigned to the client is much more greater than 20.

I think @RongtongJin means to verify the performance in some abnormal scenarios, such as restarting the broker during the process of pulling messages to see if it works normally

Got it! I will try to achieve it.

github-actions · 2024-04-12T00:08:10Z

This PR is stale because it has been open for 365 days with no activity. It will be closed in 3 days if no further activity occurs. If you wish not to mark it as stale, please leave a comment in this PR.

github-actions · 2024-04-15T02:25:28Z

This PR was closed because it has been inactive for 3 days since being marked as stale.

[ISSUE apache#6205] Support Async pull for DefaultLitePullConsumer

2eec143

buptxxb65 mentioned this pull request Mar 20, 2023

LitePullConsumer support async pull message #6205

Closed

RongtongJin reviewed Mar 21, 2023

View reviewed changes

RongtongJin reviewed Mar 22, 2023

View reviewed changes

bug fix

3029f11

guyinyou reviewed Mar 24, 2023

View reviewed changes

bug fix

44dd532

miles-ton reviewed Mar 24, 2023

View reviewed changes

Add an example for LitePullConsumer using Async pull

5813326

RongtongJin reviewed Apr 10, 2023

View reviewed changes

github-actions bot added the stale label Apr 12, 2024

github-actions bot closed this Apr 15, 2024

[ISSUE #6205] Support Async pull for DefaultLitePullConsumer #6417

[ISSUE #6205] Support Async pull for DefaultLitePullConsumer #6417

Uh oh!

Conversation

buptxxb65 commented Mar 20, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief changelog

Verifying this change

Uh oh!

codecov-commenter commented Mar 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guyinyou commented Mar 22, 2023

Uh oh!

buptxxb65 commented Mar 23, 2023

Uh oh!

github-advanced-security bot commented Mar 24, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RongtongJin commented Mar 31, 2023

Uh oh!

buptxxb65 commented Apr 2, 2023

Uh oh!

RongtongJin commented Apr 3, 2023

Uh oh!

buptxxb65 commented Apr 7, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RongtongJin commented Apr 10, 2023

Uh oh!

buptxxb65 commented Apr 12, 2023

Uh oh!

guyinyou commented Apr 12, 2023

Uh oh!

buptxxb65 commented Apr 12, 2023

Uh oh!

github-actions bot commented Apr 12, 2024

Uh oh!

github-actions bot commented Apr 15, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

buptxxb65 commented Mar 20, 2023 •

edited

Loading

codecov-commenter commented Mar 21, 2023 •

edited

Loading