[RIP-63] Queue Selection Strategy Optimization #6568

GenerousMan · 2023-04-11T07:58:07Z

Make sure set the target branch to develop

What is the purpose of the change

[RIP-63] Issue #6567 .
Optimize the proxy's and client's selection strategy for brokers when sending messages, and use multiple selection strategies as a pipeline to filter suitable queues.

Brief changelog

Client:

Modified the files in org.apache.rocketmq.client.latency, such as MQFaultStrategy and LatencyFaultTolerance. These changes implemented queues's filters, which will provide more flexible selecting strategies.
Added some switches and configs in ClientConfig, to control this strategy.
Modified the updateFaultItem() method in the sending process. This method is designed to help record the brokers' available status. Now it will also record whether brokers are reachable or not.

Proxy:

Added the MQFaultStrategy in the topicRouteService. It will provide a similar queue selection mechanism in the client.
Added some switches and configs in ProxyConfig, to control this strategy.
Implemented the selectOneByPipeline() in MessageQueueSelector. It will call the filters in mqFaultStrategy to select suitable queue while sending messages.
Modified the implement of select() in SendMessageQueueSelector, now it will call the selectOneByPipeline(), but not the selectOne() before.(PS: If the switch is not opened, it will also run as selectOne()).
Added updateFaultItem() in the ProducerProcessor's sendMessage to record brokers' status, according to sending results.

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.

…ueFilter

fuyou001 · 2023-04-11T10:36:24Z

client/src/main/java/org/apache/rocketmq/client/common/ThreadLocalIndex.java

+    public void reset() {
+        int index = Math.abs(random.nextInt());
+        if (index < 0)
+            index = 0;


the code redundant

fuyou001 · 2023-04-11T10:58:02Z

client/src/main/java/org/apache/rocketmq/client/ClientConfig.java

@@ -79,6 +89,10 @@ public class ClientConfig {
     */
    protected boolean enableStreamRequestType = false;

+    private boolean sendLatencyEnable = Boolean.parseBoolean(System.getProperty(SEND_LATENCY_ENABLE, "false"));


Would it be prudent to incorporate a warning concerning order message within the document,may be order message is disturbed

fuyou001 · 2023-04-11T11:01:37Z

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -2846,6 +2849,37 @@ public void setMessageRequestMode(final String brokerAddr, final String topic, f
            throw new MQClientException(response.getCode(), response.getRemark());
        }
    }
+    public Properties queryRemoteClientConfig(long timeoutMillis)


may be increased namesrv server load

OK, I will remove this design.

HScarb · 2023-04-12T02:01:31Z

client/src/main/java/org/apache/rocketmq/client/RemoteClientConfig.java

+ * Remoting Client Common configuration
+ */
+public class RemoteClientConfig {
+    private final static Logger log = LoggerFactory.getLogger(MQClientInstance.class);


It seems more appropriate to write it like this:
private final static Logger log = LoggerFactory.getLogger(RemoteClientConfig.class);

Yes, It should be. But I think I will remove this RemoteClientConfig...
The switches in clientConfig are enough.

GenerousMan · 2023-04-12T02:20:45Z

Some experiment results: https://docs.google.com/document/d/1uTB_5bp6YmoghGUpFabwYsMertP7IBlNCDx9r787sOU/edit?usp=sharing

client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java

…ect().

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java

# Conflicts: # client/src/main/java/org/apache/rocketmq/client/ClientConfig.java # proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java # proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java # proxy/src/test/java/org/apache/rocketmq/proxy/service/BaseServiceTest.java # proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java # proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java

# Conflicts: # proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java

# Conflicts: # client/src/main/java/org/apache/rocketmq/client/latency/MQFaultStrategy.java

# Conflicts: # client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java # proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java # proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java

xdkxlk

LGTM

xdkxlk · 2023-08-25T07:09:39Z

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java

                    return sendResultList;
-                }, this.executor);
+                }, this.executor)
+                    .exceptionally(t -> {


Use whenComplete is better. When an exception occurs, the upper layer function will get null in thenApply

Some methods need to be modified

drpmma · 2023-08-28T07:06:48Z

client/src/main/java/org/apache/rocketmq/client/latency/LatencyFaultToleranceImpl.java

-                final int i = this.randomItem.incrementAndGet() % half;
-                return tmpList.get(i).getName();
+            Collections.shuffle(tmpList);
+            //Collections.sort(tmpList);


annotation could be removed

drpmma · 2023-08-28T07:34:21Z

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java

-                }, this.executor);
+                }, this.executor)
+                    .whenComplete((result, exception) -> {
+                        endTimestamp.set(System.currentTimeMillis());


There is no need to use AtomicLong for endTimestamp, just a local variable would be fine

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java

drpmma

LGTM

drpmma

LGTM

GenerousMan added 21 commits March 28, 2023 17:28

add latency files.

a4f01fc

modify the logic of selectOneMessageQueue

96f90ca

faultStrategy modify.

63cbbac

modify updateFaultItems.

4bfe568

add remoting switch

da61ed2

start the detector

8ddf4a2

modify the config

19bb1d9

modify the config

e22b40b

prepare the proxy modification

f0a4b73

proxy's latencyTolerance done, then we need filters.

ab08b2d

fix style.

fde2dea

test the proxy

9ee6dab

finish proxy's faultStrategy.

d4216c1

remove useless files.

542369e

checkstyle.

19bd944

add exceptionally logic.

ae362da

fix unit test's bug.

472ecc9

add a unit test.

4c01506

remove useless methods.

83f55e4

Merge branch 'develop' into queueFilter

0d83652

Merge branch 'develop' of https://github.com/apache/rocketmq into que…

98516ac

…ueFilter

fuyou001 reviewed Apr 11, 2023

View reviewed changes

HScarb reviewed Apr 12, 2023

View reviewed changes

RongtongJin reviewed Apr 12, 2023

View reviewed changes

client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java Show resolved Hide resolved

GenerousMan added 5 commits April 12, 2023 11:24

refactor

ded23c3

remove useless import and methods.

cf9fe17

optimized the reset method.

3841201

still using getMaxOffset() to detect brokers' status.

ffe22e5

make BrokerFilter static, to avoid the find-bug's notification.

ee385d8

GenerousMan added 3 commits April 23, 2023 10:33

Put TopicRouteService into queueSelector, instead of passing into sel…

f1f52f9

…ect().

fix checkstyle.

27dd55f

fix the test case 'testSendNormalMessageQueueSelector'.

2610988

xdkxlk reviewed Apr 25, 2023

View reviewed changes

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java Outdated Show resolved Hide resolved

GenerousMan added 14 commits April 25, 2023 14:36

optimize the queueSelector's init function.

8535cd9

solve conflict.

1957f69

Merge branch 'develop' into queueFilter_new

27ce8f6

polish.

02c560d

Merge branch 'develop' into queueFilter_new

481ba50

# Conflicts: # proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java

fix conflict.

774f486

polish

e9334eb

polish

6b39ab5

polish

3a43bde

Merge branch 'develop' into queueFilter_new

70efe49

# Conflicts: # client/src/main/java/org/apache/rocketmq/client/latency/MQFaultStrategy.java

fix wrong modification.

196a642

make the switch false default.

90320fa

xdkxlk previously approved these changes Aug 25, 2023

View reviewed changes

xdkxlk reviewed Aug 25, 2023

View reviewed changes

use whenComplete instead of exceptionally.

ce56323

drpmma requested changes Aug 28, 2023

View reviewed changes

modify the requested changes.

1ec969b

drpmma previously approved these changes Aug 29, 2023

View reviewed changes

checkstyle.

0924aae

GenerousMan dismissed drpmma’s stale review via 0924aae August 29, 2023 03:18

fix test bug.

57279e8

drpmma approved these changes Aug 29, 2023

View reviewed changes

drpmma merged commit b028277 into apache:develop Aug 29, 2023
9 of 10 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[RIP-63] Queue Selection Strategy Optimization #6568

[RIP-63] Queue Selection Strategy Optimization #6568

GenerousMan commented Apr 11, 2023

fuyou001 Apr 11, 2023

fuyou001 Apr 11, 2023 •

edited

fuyou001 Apr 11, 2023

GenerousMan Apr 12, 2023

HScarb Apr 12, 2023

GenerousMan Apr 12, 2023

GenerousMan commented Apr 12, 2023

xdkxlk left a comment

xdkxlk Aug 25, 2023 •

edited

drpmma Aug 28, 2023

drpmma Aug 28, 2023

drpmma left a comment

drpmma left a comment

[RIP-63] Queue Selection Strategy Optimization #6568

[RIP-63] Queue Selection Strategy Optimization #6568

Conversation

GenerousMan commented Apr 11, 2023

What is the purpose of the change

Brief changelog

Client:

Proxy:

Verifying this change

fuyou001 Apr 11, 2023

Choose a reason for hiding this comment

fuyou001 Apr 11, 2023 • edited

Choose a reason for hiding this comment

fuyou001 Apr 11, 2023

Choose a reason for hiding this comment

GenerousMan Apr 12, 2023

Choose a reason for hiding this comment

HScarb Apr 12, 2023

Choose a reason for hiding this comment

GenerousMan Apr 12, 2023

Choose a reason for hiding this comment

GenerousMan commented Apr 12, 2023

xdkxlk left a comment

Choose a reason for hiding this comment

xdkxlk Aug 25, 2023 • edited

Choose a reason for hiding this comment

drpmma Aug 28, 2023

Choose a reason for hiding this comment

drpmma Aug 28, 2023

Choose a reason for hiding this comment

drpmma left a comment

Choose a reason for hiding this comment

drpmma left a comment

Choose a reason for hiding this comment

fuyou001 Apr 11, 2023 •

edited

xdkxlk Aug 25, 2023 •

edited