You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "GenerousMan (via GitHub)" <gi...@apache.org> on 2023/04/11 07:58:11 UTC
[GitHub] [rocketmq] GenerousMan opened a new pull request, #6568: [RIP-63] Queue Selection Strategy Optimization
GenerousMan opened a new pull request, #6568:
URL: https://github.com/apache/rocketmq/pull/6568
**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`.
- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) 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.
- [x] 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.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] 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](https://github.com/apache/rocketmq/tree/master/test).
- [x] 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](http://www.apache.org/licenses/#clas).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fuyou001 commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1162646629
##########
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"));
Review Comment:
Would it be prudent to incorporate a warning concerning order message within the document
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] GenerousMan commented on pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "GenerousMan (via GitHub)" <gi...@apache.org>.
GenerousMan commented on PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#issuecomment-1504447591
Some experiment results: https://docs.google.com/document/d/1uTB_5bp6YmoghGUpFabwYsMertP7IBlNCDx9r787sOU/edit?usp=sharing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fuyou001 commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1162625334
##########
client/src/main/java/org/apache/rocketmq/client/common/ThreadLocalIndex.java:
##########
@@ -33,6 +33,13 @@ public int incrementAndGet() {
return Math.abs(index & POSITIVE_MASK);
}
+ public void reset() {
+ int index = Math.abs(random.nextInt());
+ if (index < 0)
+ index = 0;
Review Comment:
the code redundant
##########
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"));
Review Comment:
Would it be prudent to incorporate a warning concerning order messaging within the document
##########
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)
Review Comment:
may be increased namesrv server load
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] xdkxlk commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "xdkxlk (via GitHub)" <gi...@apache.org>.
xdkxlk commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1305272660
##########
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java:
##########
@@ -102,11 +108,18 @@ public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSe
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) &&
tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE &&
StringUtils.isNotBlank(sendResult.getTransactionId())) {
- fillTransactionData(ctx, producerGroup, messageQueue, sendResult, messageList);
+ fillTransactionData(ctx, producerGroup, finalMessageQueue, sendResult, messageList);
}
}
+ endTimestamp.set(System.currentTimeMillis());
+ this.serviceManager.getTopicRouteService().updateFaultItem(finalMessageQueue.getBrokerName(),endTimestamp.get() - beginTimestampFirst, false, true);
return sendResultList;
- }, this.executor);
+ }, this.executor)
+ .exceptionally(t -> {
Review Comment:
Use whenComplete is better. When an exception occurs, the upper layer function will get null in thenApply
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] xdkxlk commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "xdkxlk (via GitHub)" <gi...@apache.org>.
xdkxlk commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1305272660
##########
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java:
##########
@@ -102,11 +108,18 @@ public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSe
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) &&
tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE &&
StringUtils.isNotBlank(sendResult.getTransactionId())) {
- fillTransactionData(ctx, producerGroup, messageQueue, sendResult, messageList);
+ fillTransactionData(ctx, producerGroup, finalMessageQueue, sendResult, messageList);
}
}
+ endTimestamp.set(System.currentTimeMillis());
+ this.serviceManager.getTopicRouteService().updateFaultItem(finalMessageQueue.getBrokerName(),endTimestamp.get() - beginTimestampFirst, false, true);
return sendResultList;
- }, this.executor);
+ }, this.executor)
+ .exceptionally(t -> {
Review Comment:
Use whenComplete is better. When an exception occurs, the upper layer function will get null in thApply
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] GenerousMan commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "GenerousMan (via GitHub)" <gi...@apache.org>.
GenerousMan commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1176066998
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java:
##########
@@ -154,6 +171,82 @@ public AddressableMessageQueue selectOne(boolean onlyBroker) {
return selectOneByIndex(nextIndex, onlyBroker);
}
+ public AddressableMessageQueue selectOneByPipeline(boolean onlyBroker) {
+ if (topicRouteService.getMqFaultStrategy().isSendLatencyFaultEnable() && topicRouteService != null) {
Review Comment:
Yes, I have modified as you said. You can check commits history.
Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] xdkxlk commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "xdkxlk (via GitHub)" <gi...@apache.org>.
xdkxlk commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1175944390
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java:
##########
@@ -154,6 +171,82 @@ public AddressableMessageQueue selectOne(boolean onlyBroker) {
return selectOneByIndex(nextIndex, onlyBroker);
}
+ public AddressableMessageQueue selectOneByPipeline(boolean onlyBroker) {
+ if (topicRouteService.getMqFaultStrategy().isSendLatencyFaultEnable() && topicRouteService != null) {
Review Comment:
TopicRouteService could be null if it is not passed into the constructor.
Moreover, if the constructor without topicRouteService is no longer used, we can delete it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1163486851
##########
client/src/main/java/org/apache/rocketmq/client/RemoteClientConfig.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.client;
+
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+
+/**
+ * Remoting Client Common configuration
+ */
+public class RemoteClientConfig {
+ private final static Logger log = LoggerFactory.getLogger(MQClientInstance.class);
Review Comment:
It seems more appropriate to write it like this:
`private final static Logger log = LoggerFactory.getLogger(RemoteClientConfig.class);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] codecov-commenter commented on pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#issuecomment-1510604944
## [Codecov](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#6568](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ee385d8) into [develop](https://codecov.io/gh/apache/rocketmq/commit/7dc9942fe20dfee8841221e55c904f872c994d38?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7dc9942) will **increase** coverage by `0.00%`.
> The diff coverage is `42.35%`.
```diff
@@ Coverage Diff @@
## develop #6568 +/- ##
===========================================
Coverage 43.08% 43.09%
- Complexity 8991 9017 +26
===========================================
Files 1107 1106 -1
Lines 78257 78534 +277
Branches 10201 10234 +33
===========================================
+ Hits 33718 33843 +125
- Misses 40316 40441 +125
- Partials 4223 4250 +27
```
| [Impacted Files](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...pache/rocketmq/client/common/ThreadLocalIndex.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvY29tbW9uL1RocmVhZExvY2FsSW5kZXguamF2YQ==) | `66.66% <0.00%> (-33.34%)` | :arrow_down: |
| [...tmq/namesrv/processor/DefaultRequestProcessor.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bmFtZXNydi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcm9ja2V0bXEvbmFtZXNydi9wcm9jZXNzb3IvRGVmYXVsdFJlcXVlc3RQcm9jZXNzb3IuamF2YQ==) | `61.56% <ø> (+2.55%)` | :arrow_up: |
| [...mq/client/impl/producer/DefaultMQProducerImpl.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9wcm9kdWNlci9EZWZhdWx0TVFQcm9kdWNlckltcGwuamF2YQ==) | `44.57% <20.00%> (-1.40%)` | :arrow_down: |
| [...ocketmq/proxy/service/route/TopicRouteService.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L3NlcnZpY2Uvcm91dGUvVG9waWNSb3V0ZVNlcnZpY2UuamF2YQ==) | `45.55% <23.52%> (-13.38%)` | :arrow_down: |
| [...g/apache/rocketmq/client/impl/MQClientAPIImpl.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9NUUNsaWVudEFQSUltcGwuamF2YQ==) | `23.19% <33.33%> (ø)` | |
| [...etmq/client/latency/LatencyFaultToleranceImpl.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvbGF0ZW5jeS9MYXRlbmN5RmF1bHRUb2xlcmFuY2VJbXBsLmphdmE=) | `39.37% <37.33%> (-9.99%)` | :arrow_down: |
| [...pache/rocketmq/client/latency/MQFaultStrategy.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvbGF0ZW5jeS9NUUZhdWx0U3RyYXRlZ3kuamF2YQ==) | `37.83% <42.10%> (+19.65%)` | :arrow_up: |
| [.../org/apache/rocketmq/proxy/config/ProxyConfig.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2NvbmZpZy9Qcm94eUNvbmZpZy5qYXZh) | `53.91% <44.44%> (-0.30%)` | :arrow_down: |
| [.../java/org/apache/rocketmq/client/ClientConfig.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvQ2xpZW50Q29uZmlnLmphdmE=) | `61.34% <45.45%> (-2.04%)` | :arrow_down: |
| [...rocketmq/client/impl/factory/MQClientInstance.java](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9mYWN0b3J5L01RQ2xpZW50SW5zdGFuY2UuamF2YQ==) | `46.90% <50.00%> (+<0.01%)` | :arrow_up: |
| ... and [6 more](https://codecov.io/gh/apache/rocketmq/pull/6568?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
... and [30 files with indirect coverage changes](https://codecov.io/gh/apache/rocketmq/pull/6568/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] drpmma merged pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "drpmma (via GitHub)" <gi...@apache.org>.
drpmma merged PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] GenerousMan commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "GenerousMan (via GitHub)" <gi...@apache.org>.
GenerousMan commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1163483466
##########
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)
Review Comment:
OK, I will remove this design.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "RongtongJin (via GitHub)" <gi...@apache.org>.
RongtongJin commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1163532994
##########
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java:
##########
@@ -153,6 +157,38 @@ public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook
semaphoreAsyncSendSize = new Semaphore(1024 * 1024, true);
log.info("semaphoreAsyncSendSize can not be smaller than 1M.");
}
+
+ ServiceDetector serviceDetector = new ServiceDetector() {
+ @Override
+ public boolean detect(String endpoint, long timeoutMillis) {
+ Optional<String> candidateTopic = pickTopic();
+ if (!candidateTopic.isPresent()) {
+ return false;
+ }
+ try {
+ MessageQueue mq = new MessageQueue(candidateTopic.get(), null, 0);
+ mQClientFactory.getMQClientAPIImpl()
+ .getMaxOffset(endpoint, mq, timeoutMillis);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+ };
Review Comment:
When there are many clients (such as 1 million producers or consumers, which by default probe every broker every 3 seconds), is GetMaxOffset a relatively heavy operation? Could it put pressure on the broker? Here are my two suggestions:
1.The RemotingClient's isAddressReachable method could be used to complete connection detection and check if the long connection channel is normal, which is lighter weight.
2.If it is necessary to check if the broker's execution logic is normal, could it be replaced with VIEW_BROKER_STATS_DATA? This request does not require topic or message queue related data and does not involve the store layer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] xdkxlk commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "xdkxlk (via GitHub)" <gi...@apache.org>.
xdkxlk commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1173262777
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java:
##########
@@ -154,6 +157,82 @@ public AddressableMessageQueue selectOne(boolean onlyBroker) {
return selectOneByIndex(nextIndex, onlyBroker);
}
+ public AddressableMessageQueue selectOneByPipeline(TopicRouteService topicRouteService, boolean onlyBroker) {
Review Comment:
It would be better to pass the TopicRouteService when constructing the MessageQueueSelector
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fuyou001 commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1162646629
##########
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"));
Review Comment:
Would it be prudent to incorporate a warning concerning order message within the document,may be order message disturbed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] GenerousMan commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "GenerousMan (via GitHub)" <gi...@apache.org>.
GenerousMan commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1163538597
##########
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java:
##########
@@ -153,6 +157,38 @@ public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook
semaphoreAsyncSendSize = new Semaphore(1024 * 1024, true);
log.info("semaphoreAsyncSendSize can not be smaller than 1M.");
}
+
+ ServiceDetector serviceDetector = new ServiceDetector() {
+ @Override
+ public boolean detect(String endpoint, long timeoutMillis) {
+ Optional<String> candidateTopic = pickTopic();
+ if (!candidateTopic.isPresent()) {
+ return false;
+ }
+ try {
+ MessageQueue mq = new MessageQueue(candidateTopic.get(), null, 0);
+ mQClientFactory.getMQClientAPIImpl()
+ .getMaxOffset(endpoint, mq, timeoutMillis);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+ };
Review Comment:
Very elegant solution. I will modify as you said.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] GenerousMan commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "GenerousMan (via GitHub)" <gi...@apache.org>.
GenerousMan commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1174513613
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java:
##########
@@ -154,6 +157,82 @@ public AddressableMessageQueue selectOne(boolean onlyBroker) {
return selectOneByIndex(nextIndex, onlyBroker);
}
+ public AddressableMessageQueue selectOneByPipeline(TopicRouteService topicRouteService, boolean onlyBroker) {
Review Comment:
You are right. I have modified codes as you said. You can check the commit history.
Thank you~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] drpmma commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "drpmma (via GitHub)" <gi...@apache.org>.
drpmma commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1307024124
##########
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java:
##########
@@ -102,11 +108,19 @@ public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSe
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) &&
tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE &&
StringUtils.isNotBlank(sendResult.getTransactionId())) {
- fillTransactionData(ctx, producerGroup, messageQueue, sendResult, messageList);
+ fillTransactionData(ctx, producerGroup, finalMessageQueue, sendResult, messageList);
}
}
return sendResultList;
- }, this.executor);
+ }, this.executor)
+ .whenComplete((result, exception) -> {
+ endTimestamp.set(System.currentTimeMillis());
Review Comment:
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:
##########
@@ -44,8 +47,9 @@ public class MessageQueueSelector {
private final Map<String, AddressableMessageQueue> brokerNameQueueMap = new ConcurrentHashMap<>();
private final AtomicInteger queueIndex;
private final AtomicInteger brokerIndex;
+ private TopicRouteService topicRouteService;
- public MessageQueueSelector(TopicRouteWrapper topicRouteWrapper, boolean read) {
+ public MessageQueueSelector(TopicRouteWrapper topicRouteWrapper, TopicRouteService topicRouteService, boolean read) {
Review Comment:
Why not pass MqFaultStrategy instead of TopicRouteService as TopicRouteService is not actually used in this class.
##########
client/src/main/java/org/apache/rocketmq/client/latency/LatencyFaultToleranceImpl.java:
##########
@@ -65,68 +140,99 @@ public void remove(final String name) {
@Override
public String pickOneAtLeast() {
final Enumeration<FaultItem> elements = this.faultItemTable.elements();
- List<FaultItem> tmpList = new LinkedList<>();
+ List<FaultItem> tmpList = new LinkedList<FaultItem>();
while (elements.hasMoreElements()) {
final FaultItem faultItem = elements.nextElement();
tmpList.add(faultItem);
}
+
if (!tmpList.isEmpty()) {
- Collections.sort(tmpList);
- final int half = tmpList.size() / 2;
- if (half <= 0) {
- return tmpList.get(0).getName();
- } else {
- final int i = this.randomItem.incrementAndGet() % half;
- return tmpList.get(i).getName();
+ Collections.shuffle(tmpList);
+ //Collections.sort(tmpList);
Review Comment:
annotation could be removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fuyou001 commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1162646629
##########
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"));
Review Comment:
Would it be prudent to incorporate a warning concerning order message within the document,may be order message is disturbed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] GenerousMan commented on a diff in pull request #6568: [RIP-63] Queue Selection Strategy Optimization
Posted by "GenerousMan (via GitHub)" <gi...@apache.org>.
GenerousMan commented on code in PR #6568:
URL: https://github.com/apache/rocketmq/pull/6568#discussion_r1163499810
##########
client/src/main/java/org/apache/rocketmq/client/RemoteClientConfig.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.client;
+
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+
+/**
+ * Remoting Client Common configuration
+ */
+public class RemoteClientConfig {
+ private final static Logger log = LoggerFactory.getLogger(MQClientInstance.class);
Review Comment:
Yes, It should be. But I think I will remove this RemoteClientConfig...
The switches in clientConfig are enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org