You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2022/09/19 02:11:19 UTC
[rocketmq] branch release-5.0.0 updated (8ba84c31d -> a174e0bfc)
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a change to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
from 8ba84c31d Bump DLedger to 0.3.1 (#5082)
new e11dc27c4 [ISSUE #5069] polish the startup of proxy; can specify parameters on the command line of proxy (#5083)
new afea709ac [ISSUE #5097] fix illegal reflective access (#5098)
new ac7e4cbfc Sync DLedger version in Bazel scripts (#5102)
new 02aa4758b Sync commons-lang3 version (#5103)
new 1b011c1f5 [ISSUE #5108] Bump checkstyle to 3.2.0 (#5109)
new 2aa8124be Fixed intermixing of punctuation marks in English documents (#5086)
new a174e0bfc [ISSUE #5104] Clean property PROPERTY_TIMER_DELAY_MS if not timer message #
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.licenserc.yaml | 1 +
WORKSPACE | 13 +-
.../org/apache/rocketmq/acl/common/Permission.java | 2 +-
broker/BUILD.bazel | 2 +-
.../apache/rocketmq/broker/BrokerController.java | 2 +-
.../broker/processor/AdminBrokerProcessor.java | 2 +-
.../broker/processor/PopBufferMergeService.java | 10 +-
.../broker/processor/PopReviveService.java | 2 +-
.../queue/TransactionalMessageServiceImpl.java | 8 +-
.../org/apache/rocketmq/broker/util/HookUtils.java | 3 +
.../consumer/ConsumeMessageOrderlyService.java | 2 +-
.../impl/consumer/DefaultMQPushConsumerImpl.java | 4 +-
.../org/apache/rocketmq/common/DataVersion.java | 2 +-
.../java/org/apache/rocketmq/common/UtilAll.java | 32 +--
.../rocketmq/common/filter/impl/PolishExpr.java | 4 +-
.../common/protocol/body/ConsumerRunningInfo.java | 2 +-
.../statictopic/TopicQueueMappingDetail.java | 4 +-
container/BUILD.bazel | 4 +-
controller/BUILD.bazel | 4 +-
.../impl/DefaultBrokerHeartbeatManager.java | 2 +-
distribution/bin/mqbroker | 35 +++-
distribution/bin/mqshutdown | 7 +-
distribution/conf/rmq-proxy.json | 2 +-
..._Topic_Logic_Queue_\350\256\276\350\256\241.md" | 2 +-
docs/en/Configuration_TLS.md | 8 +-
docs/en/Design_Remoting.md | 2 +-
docs/en/Design_Store.md | 2 +-
docs/en/FAQ.md | 2 +-
docs/en/Operations_Trace.md | 6 +-
docs/en/README.md | 22 +--
docs/en/Troubleshoopting.md | 4 +-
docs/en/architecture.md | 18 +-
docs/en/best_practice.md | 2 +-
.../client/java/API_Reference_DefaultMQProducer.md | 2 +-
docs/en/dledger/deploy_guide.md | 4 +-
docs/en/dledger/quick_start.md | 6 +-
docs/en/operation.md | 18 +-
.../rocketmq/logging/inner/LoggingBuilder.java | 4 +-
.../namesrv/processor/ClientRequestProcessor.java | 2 +-
.../rocketmq/promise/DefaultPromise.java | 2 +-
pom.xml | 4 +-
proxy/BUILD.bazel | 2 +
.../apache/rocketmq/proxy/CommandLineArgument.java | 39 ++--
.../org/apache/rocketmq/proxy/ProxyStartup.java | 85 +++++++-
.../rocketmq/proxy/config/Configuration.java | 28 +--
.../apache/rocketmq/proxy/config/ProxyConfig.java | 33 ++--
.../proxy/service/mqclient/MQClientAPIFactory.java | 8 +-
.../apache/rocketmq/proxy/ProxyStartupTest.java | 220 +++++++++++++++++++++
.../service/message/LocalMessageServiceTest.java | 2 +-
.../org.mockito.plugins.MockMaker | 1 +
store/BUILD.bazel | 42 ++--
store/libs/dledger-0.2.8-SNAPSHOT.jar | Bin 186014 -> 0 bytes
.../apache/rocketmq/store/ha/DefaultHAService.java | 4 +-
.../org/apache/rocketmq/store/index/IndexFile.java | 6 +-
.../rocketmq/store/queue/BatchConsumeQueue.java | 2 +-
.../apache/rocketmq/test/grpc/v2/GrpcBaseIT.java | 2 +-
.../tools/admin/DefaultMQAdminExtImpl.java | 2 +-
.../rocketmq/tools/command/MQAdminStartup.java | 2 +-
.../tools/command/stats/StatsAllSubCommand.java | 6 +-
.../command/topic/UpdateStaticTopicSubCommand.java | 2 +-
60 files changed, 550 insertions(+), 195 deletions(-)
copy tools/src/main/java/org/apache/rocketmq/tools/monitor/MonitorConfig.java => proxy/src/main/java/org/apache/rocketmq/proxy/CommandLineArgument.java (57%)
create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java
create mode 100644 proxy/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
delete mode 100644 store/libs/dledger-0.2.8-SNAPSHOT.jar
[rocketmq] 01/07: [ISSUE #5069] polish the startup of proxy; can specify parameters on the command line of proxy (#5083)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit e11dc27c495740abdc2c561acb3f5b700259a9af
Author: lk <ka...@alibaba-inc.com>
AuthorDate: Fri Sep 16 17:31:27 2022 +0800
[ISSUE #5069] polish the startup of proxy; can specify parameters on the command line of proxy (#5083)
undefined
---
.licenserc.yaml | 1 +
distribution/bin/mqbroker | 35 +++-
distribution/bin/mqshutdown | 7 +-
distribution/conf/rmq-proxy.json | 2 +-
proxy/BUILD.bazel | 2 +
.../apache/rocketmq/proxy/CommandLineArgument.java | 56 ++++++
.../org/apache/rocketmq/proxy/ProxyStartup.java | 85 +++++++-
.../rocketmq/proxy/config/Configuration.java | 28 +--
.../apache/rocketmq/proxy/config/ProxyConfig.java | 33 ++--
.../proxy/service/mqclient/MQClientAPIFactory.java | 8 +-
.../apache/rocketmq/proxy/ProxyStartupTest.java | 220 +++++++++++++++++++++
.../service/message/LocalMessageServiceTest.java | 2 +-
.../org.mockito.plugins.MockMaker | 1 +
.../apache/rocketmq/test/grpc/v2/GrpcBaseIT.java | 2 +-
14 files changed, 438 insertions(+), 44 deletions(-)
diff --git a/.licenserc.yaml b/.licenserc.yaml
index 51741f903..0d732fa04 100644
--- a/.licenserc.yaml
+++ b/.licenserc.yaml
@@ -35,6 +35,7 @@ header:
- '*/src/test/resources/META-INF/service/*'
- '*/src/main/resources/META-INF/service/*'
- '*/src/test/resources/rmq-proxy-home/conf/rmq-proxy.json'
+ - '*/src/test/resources/mockito-extensions/*'
- '**/target/**'
- '**/*.iml'
- 'docs/**'
diff --git a/distribution/bin/mqbroker b/distribution/bin/mqbroker
index 6a79c392e..17e39f07c 100644
--- a/distribution/bin/mqbroker
+++ b/distribution/bin/mqbroker
@@ -42,4 +42,37 @@ fi
export ROCKETMQ_HOME
-sh ${ROCKETMQ_HOME}/bin/runbroker.sh org.apache.rocketmq.broker.BrokerStartup $@
+other_args=" "
+enable_proxy=false
+
+while [[ $# -gt 0 ]]; do
+ case $1 in
+ --enable-proxy)
+ enable_proxy=true
+ shift
+ ;;
+ -c|--configFile)
+ broker_config="$2"
+ shift
+ shift
+ ;;
+ *)
+ other_args=${other_args}" "${1}
+ shift
+ ;;
+ esac
+done
+
+if [ "$enable_proxy" = true ]; then
+ args_for_proxy=$other_args" -pm local"
+ if [ "$broker_config" != "" ]; then
+ args_for_proxy=${args_for_proxy}" -bc "${broker_config}
+ fi
+ sh ${ROCKETMQ_HOME}/bin/runserver.sh org.apache.rocketmq.proxy.ProxyStartup ${args_for_proxy}
+else
+ args_for_broker=$other_args
+ if [ "$broker_config" != "" ]; then
+ args_for_broker=${args_for_broker}" -c "${broker_config}
+ fi
+ sh ${ROCKETMQ_HOME}/bin/runbroker.sh org.apache.rocketmq.broker.BrokerStartup ${args_for_broker}
+fi
\ No newline at end of file
diff --git a/distribution/bin/mqshutdown b/distribution/bin/mqshutdown
index 6006cfeb1..df2da0c31 100644
--- a/distribution/bin/mqshutdown
+++ b/distribution/bin/mqshutdown
@@ -17,7 +17,12 @@
case $1 in
broker)
-
+ pid=`ps ax | grep -i 'org.apache.rocketmq.proxy.ProxyStartup' | grep '\-pm local' |grep java | grep -v grep | awk '{print $1}'`
+ if [ "$pid" != "" ] ; then
+ echo "The mqbroker with proxy enable is running(${pid})..."
+ kill ${pid}
+ echo "Send shutdown request to mqbroker with proxy enable OK(${pid})"
+ fi
pid=`ps ax | grep -i 'org.apache.rocketmq.broker.BrokerStartup' |grep java | grep -v grep | awk '{print $1}'`
if [ -z "$pid" ] ; then
echo "No mqbroker running."
diff --git a/distribution/conf/rmq-proxy.json b/distribution/conf/rmq-proxy.json
index 077404aaa..8e92bb18e 100644
--- a/distribution/conf/rmq-proxy.json
+++ b/distribution/conf/rmq-proxy.json
@@ -1,3 +1,3 @@
{
-
+ "rocketMQClusterName": "DefaultCluster"
}
\ No newline at end of file
diff --git a/proxy/BUILD.bazel b/proxy/BUILD.bazel
index eba692157..420267ec0 100644
--- a/proxy/BUILD.bazel
+++ b/proxy/BUILD.bazel
@@ -26,6 +26,7 @@ java_library(
"//common",
"//client",
"//broker",
+ "//srvutil",
"//acl",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:org_apache_commons_commons_lang3",
@@ -50,6 +51,7 @@ java_library(
"@maven//:ch_qos_logback_logback_classic",
"@maven//:com_google_code_findbugs_jsr305",
"@maven//:org_checkerframework_checker_qual",
+ "@maven//:commons_cli_commons_cli",
],
)
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/CommandLineArgument.java b/proxy/src/main/java/org/apache/rocketmq/proxy/CommandLineArgument.java
new file mode 100644
index 000000000..0499f2659
--- /dev/null
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/CommandLineArgument.java
@@ -0,0 +1,56 @@
+/*
+ * 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.proxy;
+
+public class CommandLineArgument {
+ private String namesrvAddr;
+ private String brokerConfigPath;
+ private String proxyConfigPath;
+ private String proxyMode;
+
+ public String getNamesrvAddr() {
+ return namesrvAddr;
+ }
+
+ public void setNamesrvAddr(String namesrvAddr) {
+ this.namesrvAddr = namesrvAddr;
+ }
+
+ public String getBrokerConfigPath() {
+ return brokerConfigPath;
+ }
+
+ public void setBrokerConfigPath(String brokerConfigPath) {
+ this.brokerConfigPath = brokerConfigPath;
+ }
+
+ public String getProxyConfigPath() {
+ return proxyConfigPath;
+ }
+
+ public void setProxyConfigPath(String proxyConfigPath) {
+ this.proxyConfigPath = proxyConfigPath;
+ }
+
+ public String getProxyMode() {
+ return proxyMode;
+ }
+
+ public void setProxyMode(String proxyMode) {
+ this.proxyMode = proxyMode;
+ }
+}
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
index 9be0abe20..f605df0bf 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
@@ -20,11 +20,17 @@ package org.apache.rocketmq.proxy;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
+import com.google.common.collect.Lists;
import io.grpc.protobuf.services.ChannelzService;
import io.grpc.protobuf.services.ProtoReflectionService;
import java.util.Date;
+import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerStartup;
@@ -36,6 +42,7 @@ import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown;
import org.apache.rocketmq.proxy.common.StartAndShutdown;
+import org.apache.rocketmq.proxy.config.Configuration;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.grpc.GrpcServer;
@@ -43,6 +50,8 @@ import org.apache.rocketmq.proxy.grpc.GrpcServerBuilder;
import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingApplication;
import org.apache.rocketmq.proxy.processor.DefaultMessagingProcessor;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.srvutil.ServerUtil;
import org.slf4j.LoggerFactory;
public class ProxyStartup {
@@ -58,9 +67,9 @@ public class ProxyStartup {
public static void main(String[] args) {
try {
- ConfigurationManager.initEnv();
- initLogger();
- ConfigurationManager.intConfig();
+ // parse argument from command line
+ CommandLineArgument commandLineArgument = parseCommandLineArgument(args);
+ initLogAndConfiguration(commandLineArgument);
// init thread pool monitor for proxy.
initThreadPoolMonitor();
@@ -100,7 +109,59 @@ public class ProxyStartup {
log.info(new Date() + " rocketmq-proxy startup successfully");
}
- private static MessagingProcessor createMessagingProcessor() {
+ protected static void initLogAndConfiguration(CommandLineArgument commandLineArgument) throws Exception {
+ if (StringUtils.isNotBlank(commandLineArgument.getProxyConfigPath())) {
+ System.setProperty(Configuration.CONFIG_PATH_PROPERTY, commandLineArgument.getProxyConfigPath());
+ }
+ ConfigurationManager.initEnv();
+ initLogger();
+ ConfigurationManager.intConfig();
+ setConfigFromCommandLineArgument(commandLineArgument);
+ }
+
+ protected static CommandLineArgument parseCommandLineArgument(String[] args) {
+ CommandLine commandLine = ServerUtil.parseCmdLine("mqproxy", args,
+ buildCommandlineOptions(), new DefaultParser());
+ if (commandLine == null) {
+ throw new RuntimeException("parse command line argument failed");
+ }
+
+ CommandLineArgument commandLineArgument = new CommandLineArgument();
+ MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), commandLineArgument);
+ return commandLineArgument;
+ }
+
+ private static Options buildCommandlineOptions() {
+ Options options = ServerUtil.buildCommandlineOptions(new Options());
+
+ Option opt = new Option("bc", "brokerConfigPath", true, "Broker config file path for local mode");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("pc", "proxyConfigPath", true, "Proxy config file path");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ opt = new Option("pm", "proxyMode", true, "Proxy run in local or cluster mode");
+ opt.setRequired(false);
+ options.addOption(opt);
+
+ return options;
+ }
+
+ private static void setConfigFromCommandLineArgument(CommandLineArgument commandLineArgument) {
+ if (StringUtils.isNotBlank(commandLineArgument.getNamesrvAddr())) {
+ ConfigurationManager.getProxyConfig().setNamesrvAddr(commandLineArgument.getNamesrvAddr());
+ }
+ if (StringUtils.isNotBlank(commandLineArgument.getBrokerConfigPath())) {
+ ConfigurationManager.getProxyConfig().setBrokerConfigPath(commandLineArgument.getBrokerConfigPath());
+ }
+ if (StringUtils.isNotBlank(commandLineArgument.getProxyMode())) {
+ ConfigurationManager.getProxyConfig().setProxyMode(commandLineArgument.getProxyMode());
+ }
+ }
+
+ protected static MessagingProcessor createMessagingProcessor() {
String proxyModeStr = ConfigurationManager.getProxyConfig().getProxyMode();
MessagingProcessor messagingProcessor;
@@ -112,6 +173,12 @@ public class ProxyStartup {
@Override
public void start() throws Exception {
brokerController.start();
+ String tip = "The broker[" + brokerController.getBrokerConfig().getBrokerName() + ", "
+ + brokerController.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
+ if (null != brokerController.getBrokerConfig().getNamesrvAddr()) {
+ tip += " and name server is " + brokerController.getBrokerConfig().getNamesrvAddr();
+ }
+ log.info(tip);
}
@Override
@@ -134,8 +201,14 @@ public class ProxyStartup {
return application;
}
- private static BrokerController createBrokerController() {
- String[] brokerStartupArgs = new String[] {"-c", ConfigurationManager.getProxyConfig().getBrokerConfigPath()};
+ protected static BrokerController createBrokerController() {
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ List<String> brokerStartupArgList = Lists.newArrayList("-c", config.getBrokerConfigPath());
+ if (StringUtils.isNotBlank(config.getNamesrvAddr())) {
+ brokerStartupArgList.add("-n");
+ brokerStartupArgList.add(config.getNamesrvAddr());
+ }
+ String[] brokerStartupArgs = brokerStartupArgList.toArray(new String[0]);
return BrokerStartup.createBrokerController(brokerStartupArgs);
}
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/config/Configuration.java b/proxy/src/main/java/org/apache/rocketmq/proxy/config/Configuration.java
index 59078c712..9c1ff811b 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/config/Configuration.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/config/Configuration.java
@@ -26,6 +26,7 @@ import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.constant.LoggerName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,37 +34,38 @@ import org.slf4j.LoggerFactory;
public class Configuration {
private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
private final AtomicReference<ProxyConfig> proxyConfigReference = new AtomicReference<>();
+ public static final String CONFIG_PATH_PROPERTY = "com.rocketmq.proxy.configPath";
public void init() throws Exception {
- String proxyConfigData = loadJsonConfig(ProxyConfig.CONFIG_FILE_NAME);
- if (null == proxyConfigData) {
- throw new RuntimeException(String.format("load configuration from file: %s error.", ProxyConfig.CONFIG_FILE_NAME));
- }
+ String proxyConfigData = loadJsonConfig();
ProxyConfig proxyConfig = JSON.parseObject(proxyConfigData, ProxyConfig.class);
proxyConfig.initData();
setProxyConfig(proxyConfig);
}
- public static String loadJsonConfig(String configFileName) throws Exception {
- final String testResource = "rmq-proxy-home/conf/" + configFileName;
- try (InputStream inputStream = Configuration.class.getClassLoader().getResourceAsStream(testResource)) {
- if (null != inputStream) {
- return CharStreams.toString(new InputStreamReader(inputStream, Charsets.UTF_8));
+ public static String loadJsonConfig() throws Exception {
+ String configFileName = ProxyConfig.DEFAULT_CONFIG_FILE_NAME;
+ String filePath = System.getProperty(CONFIG_PATH_PROPERTY);
+ if (StringUtils.isBlank(filePath)) {
+ final String testResource = "rmq-proxy-home/conf/" + configFileName;
+ try (InputStream inputStream = Configuration.class.getClassLoader().getResourceAsStream(testResource)) {
+ if (null != inputStream) {
+ return CharStreams.toString(new InputStreamReader(inputStream, Charsets.UTF_8));
+ }
}
+ filePath = new File(ConfigurationManager.getProxyHome() + File.separator + "conf", configFileName).toString();
}
- String filePath = new File(ConfigurationManager.getProxyHome() + File.separator + "conf", configFileName).toString();
-
File file = new File(filePath);
if (!file.exists()) {
log.warn("the config file {} not exist", filePath);
- return null;
+ throw new RuntimeException(String.format("the config file %s not exist", filePath));
}
long fileLength = file.length();
if (fileLength <= 0) {
log.warn("the config file {} length is zero", filePath);
- return null;
+ throw new RuntimeException(String.format("the config file %s length is zero", filePath));
}
return new String(Files.readAllBytes(file.toPath()), StandardCharsets.UTF_8);
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
index 5a605f28b..00a6cc35e 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.proxy.ProxyMode;
import org.slf4j.Logger;
@@ -32,7 +33,7 @@ import org.slf4j.LoggerFactory;
public class ProxyConfig implements ConfigFile {
private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
- public final static String CONFIG_FILE_NAME = "rmq-proxy.json";
+ public final static String DEFAULT_CONFIG_FILE_NAME = "rmq-proxy.json";
private static final int PROCESSOR_NUMBER = Runtime.getRuntime().availableProcessors();
private String rocketMQClusterName = "";
@@ -44,9 +45,9 @@ public class ProxyConfig implements ConfigFile {
private long printJstackInMillis = Duration.ofSeconds(60).toMillis();
private long printThreadPoolStatusInMillis = Duration.ofSeconds(3).toMillis();
- private String nameSrvAddr = "";
- private String nameSrvDomain = "";
- private String nameSrvDomainSubgroup = "";
+ private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
+ private String namesrvDomain = "";
+ private String namesrvDomainSubgroup = "";
/**
* gRPC
*/
@@ -234,28 +235,28 @@ public class ProxyConfig implements ConfigFile {
this.printThreadPoolStatusInMillis = printThreadPoolStatusInMillis;
}
- public String getNameSrvAddr() {
- return nameSrvAddr;
+ public String getNamesrvAddr() {
+ return namesrvAddr;
}
- public void setNameSrvAddr(String nameSrvAddr) {
- this.nameSrvAddr = nameSrvAddr;
+ public void setNamesrvAddr(String namesrvAddr) {
+ this.namesrvAddr = namesrvAddr;
}
- public String getNameSrvDomain() {
- return nameSrvDomain;
+ public String getNamesrvDomain() {
+ return namesrvDomain;
}
- public void setNameSrvDomain(String nameSrvDomain) {
- this.nameSrvDomain = nameSrvDomain;
+ public void setNamesrvDomain(String namesrvDomain) {
+ this.namesrvDomain = namesrvDomain;
}
- public String getNameSrvDomainSubgroup() {
- return nameSrvDomainSubgroup;
+ public String getNamesrvDomainSubgroup() {
+ return namesrvDomainSubgroup;
}
- public void setNameSrvDomainSubgroup(String nameSrvDomainSubgroup) {
- this.nameSrvDomainSubgroup = nameSrvDomainSubgroup;
+ public void setNamesrvDomainSubgroup(String namesrvDomainSubgroup) {
+ this.namesrvDomainSubgroup = namesrvDomainSubgroup;
}
public String getProxyMode() {
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIFactory.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIFactory.java
index 0b813ae60..9d7db6cf7 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIFactory.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIFactory.java
@@ -54,11 +54,11 @@ public class MQClientAPIFactory implements StartAndShutdown {
protected void init() {
System.setProperty(ClientConfig.SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "false");
ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
- if (StringUtils.isEmpty(proxyConfig.getNameSrvDomain())) {
- System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, proxyConfig.getNameSrvAddr());
+ if (StringUtils.isEmpty(proxyConfig.getNamesrvDomain())) {
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, proxyConfig.getNamesrvAddr());
} else {
- System.setProperty("rocketmq.namesrv.domain", proxyConfig.getNameSrvDomain());
- System.setProperty("rocketmq.namesrv.domain.subgroup", proxyConfig.getNameSrvDomainSubgroup());
+ System.setProperty("rocketmq.namesrv.domain", proxyConfig.getNamesrvDomain());
+ System.setProperty("rocketmq.namesrv.domain.subgroup", proxyConfig.getNamesrvDomainSubgroup());
}
}
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java
new file mode 100644
index 000000000..6adf7f3fb
--- /dev/null
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/ProxyStartupTest.java
@@ -0,0 +1,220 @@
+/*
+ * 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.proxy;
+
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.BrokerStartup;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.proxy.config.Configuration;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.apache.rocketmq.proxy.processor.DefaultMessagingProcessor;
+import org.assertj.core.util.Strings;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+
+import static org.apache.rocketmq.proxy.config.ConfigurationManager.RMQ_PROXY_HOME;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+
+public class ProxyStartupTest {
+
+ public static String mockProxyHome = "/mock/rmq/proxy/home";
+
+ @Before
+ public void before() throws Throwable {
+ URL mockProxyHomeURL = getClass().getClassLoader().getResource("rmq-proxy-home");
+ if (mockProxyHomeURL != null) {
+ mockProxyHome = mockProxyHomeURL.toURI().getPath();
+ }
+
+ if (!Strings.isNullOrEmpty(mockProxyHome)) {
+ System.setProperty(RMQ_PROXY_HOME, mockProxyHome);
+ }
+ }
+
+ @After
+ public void after() {
+ System.clearProperty(RMQ_PROXY_HOME);
+ System.clearProperty(MixAll.NAMESRV_ADDR_PROPERTY);
+ System.clearProperty(Configuration.CONFIG_PATH_PROPERTY);
+ System.clearProperty(ClientLogger.CLIENT_LOG_USESLF4J);
+ }
+
+ @Test
+ public void testParseAndInitCommandLineArgument() throws Exception {
+ Path configFilePath = Files.createTempFile("testParseAndInitCommandLineArgument", ".json");
+ String configData = "{}";
+ Files.write(configFilePath, configData.getBytes(StandardCharsets.UTF_8));
+
+ String brokerConfigPath = "brokerConfigPath";
+ String proxyConfigPath = configFilePath.toAbsolutePath().toString();
+ String proxyMode = "LOCAL";
+ String namesrvAddr = "namesrvAddr";
+ CommandLineArgument commandLineArgument = ProxyStartup.parseCommandLineArgument(new String[] {
+ "-bc", brokerConfigPath,
+ "-pc", proxyConfigPath,
+ "-pm", proxyMode,
+ "-n", namesrvAddr
+ });
+
+ assertEquals(brokerConfigPath, commandLineArgument.getBrokerConfigPath());
+ assertEquals(proxyConfigPath, commandLineArgument.getProxyConfigPath());
+ assertEquals(proxyMode, commandLineArgument.getProxyMode());
+ assertEquals(namesrvAddr, commandLineArgument.getNamesrvAddr());
+
+ ProxyStartup.initLogAndConfiguration(commandLineArgument);
+
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ assertEquals(brokerConfigPath, config.getBrokerConfigPath());
+ assertEquals(proxyMode, config.getProxyMode());
+ assertEquals(namesrvAddr, config.getNamesrvAddr());
+ }
+
+ @Test
+ public void testLocalModeWithNameSrvAddrByProperty() throws Exception {
+ String namesrvAddr = "namesrvAddr";
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, namesrvAddr);
+ CommandLineArgument commandLineArgument = ProxyStartup.parseCommandLineArgument(new String[] {
+ "-pm", "local"
+ });
+ ProxyStartup.initLogAndConfiguration(commandLineArgument);
+
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ assertEquals(namesrvAddr, config.getNamesrvAddr());
+
+ validateBrokerCreateArgsWithNamsrvAddr(config, namesrvAddr);
+ }
+
+ private void validateBrokerCreateArgsWithNamsrvAddr(ProxyConfig config, String namesrvAddr) {
+ try (MockedStatic<BrokerStartup> brokerStartupMocked = mockStatic(BrokerStartup.class);
+ MockedStatic<DefaultMessagingProcessor> messagingProcessorMocked = mockStatic(DefaultMessagingProcessor.class)) {
+ ArgumentCaptor<Object> args = ArgumentCaptor.forClass(Object.class);
+ brokerStartupMocked.when(() -> BrokerStartup.createBrokerController((String[]) args.capture()))
+ .thenReturn(mock(BrokerController.class));
+ messagingProcessorMocked.when(() -> DefaultMessagingProcessor.createForLocalMode(any(), any()))
+ .thenReturn(mock(DefaultMessagingProcessor.class));
+
+ ProxyStartup.createMessagingProcessor();
+ String[] passArgs = (String[]) args.getValue();
+ assertEquals("-c", passArgs[0]);
+ assertEquals(config.getBrokerConfigPath(), passArgs[1]);
+ assertEquals("-n", passArgs[2]);
+ assertEquals(namesrvAddr, passArgs[3]);
+ assertEquals(4, passArgs.length);
+ }
+ }
+
+ @Test
+ public void testLocalModeWithNameSrvAddrByConfigFile() throws Exception {
+ String namesrvAddr = "namesrvAddr";
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, "foo");
+ Path configFilePath = Files.createTempFile("testLocalModeWithNameSrvAddrByConfigFile", ".json");
+ String configData = "{\n" +
+ " \"namesrvAddr\": \"namesrvAddr\"\n" +
+ "}";
+ Files.write(configFilePath, configData.getBytes(StandardCharsets.UTF_8));
+
+ CommandLineArgument commandLineArgument = ProxyStartup.parseCommandLineArgument(new String[] {
+ "-pm", "local",
+ "-pc", configFilePath.toAbsolutePath().toString()
+ });
+ ProxyStartup.initLogAndConfiguration(commandLineArgument);
+
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ assertEquals(namesrvAddr, config.getNamesrvAddr());
+
+ validateBrokerCreateArgsWithNamsrvAddr(config, namesrvAddr);
+ }
+
+ @Test
+ public void testLocalModeWithNameSrvAddrByCommandLine() throws Exception {
+ String namesrvAddr = "namesrvAddr";
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, "foo");
+ Path configFilePath = Files.createTempFile("testLocalModeWithNameSrvAddrByCommandLine", ".json");
+ String configData = "{\n" +
+ " \"namesrvAddr\": \"foo\"\n" +
+ "}";
+ Files.write(configFilePath, configData.getBytes(StandardCharsets.UTF_8));
+
+ CommandLineArgument commandLineArgument = ProxyStartup.parseCommandLineArgument(new String[] {
+ "-pm", "local",
+ "-pc", configFilePath.toAbsolutePath().toString(),
+ "-n", namesrvAddr
+ });
+ ProxyStartup.initLogAndConfiguration(commandLineArgument);
+
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ assertEquals(namesrvAddr, config.getNamesrvAddr());
+
+ validateBrokerCreateArgsWithNamsrvAddr(config, namesrvAddr);
+ }
+
+ @Test
+ public void testLocalModeWithAllArgs() throws Exception {
+ String namesrvAddr = "namesrvAddr";
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, "foo");
+ Path configFilePath = Files.createTempFile("testLocalMode", ".json");
+ String configData = "{\n" +
+ " \"namesrvAddr\": \"foo\"\n" +
+ "}";
+ Files.write(configFilePath, configData.getBytes(StandardCharsets.UTF_8));
+ Path brokerConfigFilePath = Files.createTempFile("testLocalModeBrokerConfig", ".json");
+
+ CommandLineArgument commandLineArgument = ProxyStartup.parseCommandLineArgument(new String[] {
+ "-pm", "local",
+ "-pc", configFilePath.toAbsolutePath().toString(),
+ "-n", namesrvAddr,
+ "-bc", brokerConfigFilePath.toAbsolutePath().toString()
+ });
+ ProxyStartup.initLogAndConfiguration(commandLineArgument);
+
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ assertEquals(namesrvAddr, config.getNamesrvAddr());
+ assertEquals(brokerConfigFilePath.toAbsolutePath().toString(), config.getBrokerConfigPath());
+
+ validateBrokerCreateArgsWithNamsrvAddr(config, namesrvAddr);
+ }
+
+ @Test
+ public void testClusterMode() throws Exception {
+ CommandLineArgument commandLineArgument = ProxyStartup.parseCommandLineArgument(new String[] {
+ "-pm", "cluster"
+ });
+ ProxyStartup.initLogAndConfiguration(commandLineArgument);
+
+ try (MockedStatic<DefaultMessagingProcessor> messagingProcessorMocked = mockStatic(DefaultMessagingProcessor.class)) {
+ DefaultMessagingProcessor processor = mock(DefaultMessagingProcessor.class);
+ messagingProcessorMocked.when(DefaultMessagingProcessor::createForClusterMode)
+ .thenReturn(processor);
+
+ assertSame(processor, ProxyStartup.createMessagingProcessor());
+ }
+ }
+}
\ No newline at end of file
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
index e3f6edb99..e46464a6d 100644
--- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/LocalMessageServiceTest.java
@@ -114,7 +114,7 @@ public class LocalMessageServiceTest extends InitConfigAndLoggerTest {
@Before
public void setUp() throws Throwable {
super.before();
- ConfigurationManager.getProxyConfig().setNameSrvAddr("1.1.1.1");
+ ConfigurationManager.getProxyConfig().setNamesrvAddr("1.1.1.1");
channelManager = new ChannelManager();
Mockito.when(brokerControllerMock.getSendMessageProcessor()).thenReturn(sendMessageProcessorMock);
Mockito.when(brokerControllerMock.getPopMessageProcessor()).thenReturn(popMessageProcessorMock);
diff --git a/proxy/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/proxy/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
new file mode 100644
index 000000000..ca6ee9cea
--- /dev/null
+++ b/proxy/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
@@ -0,0 +1 @@
+mock-maker-inline
\ No newline at end of file
diff --git a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java
index d88b6b1c0..586149cd1 100644
--- a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT.java
@@ -152,7 +152,7 @@ public class GrpcBaseIT extends BaseConf {
ConfigurationManager.initEnv();
ConfigurationManager.intConfig();
- ConfigurationManager.getProxyConfig().setNameSrvAddr(nsAddr);
+ ConfigurationManager.getProxyConfig().setNamesrvAddr(nsAddr);
// Set LongPollingReserveTimeInMillis to 500ms to reserve more time for IT
ConfigurationManager.getProxyConfig().setLongPollingReserveTimeInMillis(500);
ConfigurationManager.getProxyConfig().setRocketMQClusterName(brokerController1.getBrokerConfig().getBrokerClusterName());
[rocketmq] 05/07: [ISSUE #5108] Bump checkstyle to 3.2.0 (#5109)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit 1b011c1f504447e6d6f1f9901d3c3fa2d3579a80
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Sun Sep 18 11:17:36 2022 +0800
[ISSUE #5108] Bump checkstyle to 3.2.0 (#5109)
* Bump checkstyle to 3.2.0
* Fix checkstyle
---
.../main/java/org/apache/rocketmq/acl/common/Permission.java | 2 +-
.../main/java/org/apache/rocketmq/broker/BrokerController.java | 2 +-
.../apache/rocketmq/broker/processor/AdminBrokerProcessor.java | 2 +-
.../rocketmq/broker/processor/PopBufferMergeService.java | 10 +++++-----
.../org/apache/rocketmq/broker/processor/PopReviveService.java | 2 +-
.../transaction/queue/TransactionalMessageServiceImpl.java | 8 ++++----
.../client/impl/consumer/ConsumeMessageOrderlyService.java | 2 +-
.../client/impl/consumer/DefaultMQPushConsumerImpl.java | 4 ++--
.../src/main/java/org/apache/rocketmq/common/DataVersion.java | 2 +-
.../org/apache/rocketmq/common/filter/impl/PolishExpr.java | 4 ++--
.../rocketmq/common/protocol/body/ConsumerRunningInfo.java | 2 +-
.../rocketmq/common/statictopic/TopicQueueMappingDetail.java | 4 ++--
.../controller/impl/DefaultBrokerHeartbeatManager.java | 2 +-
.../java/org/apache/rocketmq/logging/inner/LoggingBuilder.java | 4 ++--
.../rocketmq/namesrv/processor/ClientRequestProcessor.java | 2 +-
.../java/io/openmessaging/rocketmq/promise/DefaultPromise.java | 2 +-
pom.xml | 2 +-
.../java/org/apache/rocketmq/store/ha/DefaultHAService.java | 4 ++--
.../main/java/org/apache/rocketmq/store/index/IndexFile.java | 6 +++---
.../org/apache/rocketmq/store/queue/BatchConsumeQueue.java | 2 +-
.../org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java | 2 +-
.../java/org/apache/rocketmq/tools/command/MQAdminStartup.java | 2 +-
.../rocketmq/tools/command/stats/StatsAllSubCommand.java | 6 +++---
.../tools/command/topic/UpdateStaticTopicSubCommand.java | 2 +-
24 files changed, 40 insertions(+), 40 deletions(-)
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java b/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java
index dadcaa304..7213cf72d 100644
--- a/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java
@@ -50,7 +50,7 @@ public class Permission {
return false;
}
if ((neededPerm & ANY) > 0) {
- return ((ownedPerm & PUB) > 0) || ((ownedPerm & SUB) > 0);
+ return (ownedPerm & PUB) > 0 || (ownedPerm & SUB) > 0;
}
return (neededPerm & ownedPerm) > 0;
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 8fe80e513..e0262a105 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -1510,7 +1510,7 @@ public class BrokerController {
this.shouldStartTime = System.currentTimeMillis() + messageStoreConfig.getDisappearTimeAfterStart();
- if ((messageStoreConfig.getTotalReplicas() > 1 && this.brokerConfig.isEnableSlaveActingMaster()) || this.brokerConfig.isEnableControllerMode()) {
+ if (messageStoreConfig.getTotalReplicas() > 1 && this.brokerConfig.isEnableSlaveActingMaster() || this.brokerConfig.isEnableControllerMode()) {
isIsolated = true;
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index cba1622fd..7369e296c 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -842,7 +842,7 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
}
SearchOffsetResponseHeader offsetResponseHeader = (SearchOffsetResponseHeader) rpcResponse.getHeader();
if (offsetResponseHeader.getOffset() < 0
- || (item.checkIfEndOffsetDecided() && offsetResponseHeader.getOffset() >= item.getEndOffset())) {
+ || item.checkIfEndOffsetDecided() && offsetResponseHeader.getOffset() >= item.getEndOffset()) {
continue;
} else {
offset = item.computeStaticQueueOffsetStrictly(offsetResponseHeader.getOffset());
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
index a8f93c22b..3a1d8baea 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
@@ -137,8 +137,8 @@ public class PopBufferMergeService extends ServiceThread {
// 1. just offset & stored, not processed by scan
// 2. ck is buffer(acked)
// 3. ck is buffer(not all acked), all ak are stored and ck is stored
- if ((pointWrapper.isJustOffset() && pointWrapper.isCkStored()) || isCkDone(pointWrapper)
- || (isCkDoneForFinish(pointWrapper) && pointWrapper.isCkStored())) {
+ if (pointWrapper.isJustOffset() && pointWrapper.isCkStored() || isCkDone(pointWrapper)
+ || isCkDoneForFinish(pointWrapper) && pointWrapper.isCkStored()) {
if (commitOffset(pointWrapper)) {
queue.poll();
} else {
@@ -218,8 +218,8 @@ public class PopBufferMergeService extends ServiceThread {
PopCheckPointWrapper pointWrapper = entry.getValue();
// just process offset(already stored at pull thread), or buffer ck(not stored and ack finish)
- if ((pointWrapper.isJustOffset() && pointWrapper.isCkStored()) || isCkDone(pointWrapper)
- || (isCkDoneForFinish(pointWrapper) && pointWrapper.isCkStored())) {
+ if (pointWrapper.isJustOffset() && pointWrapper.isCkStored() || isCkDone(pointWrapper)
+ || isCkDoneForFinish(pointWrapper) && pointWrapper.isCkStored()) {
if (brokerController.getBrokerConfig().isEnablePopLog()) {
POP_LOGGER.info("[PopBuffer]ck done, {}", pointWrapper);
}
@@ -231,7 +231,7 @@ public class PopBufferMergeService extends ServiceThread {
PopCheckPoint point = pointWrapper.getCk();
long now = System.currentTimeMillis();
- boolean removeCk = this.serving ? false : true;
+ boolean removeCk = !this.serving;
// ck will be timeout
if (point.getReviveTime() - now < brokerController.getBrokerConfig().getPopCkStayBufferTimeOut()) {
removeCk = true;
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
index 8708b48a6..7eefe5876 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
@@ -173,7 +173,7 @@ public class PopReviveService extends ServiceThread {
private boolean reachTail(PullResult pullResult, long offset) {
return pullResult.getPullStatus() == PullStatus.NO_NEW_MSG
- || (pullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL && offset == pullResult.getMaxOffset());
+ || pullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL && offset == pullResult.getMaxOffset();
}
private MessageExt getBizMessage(String topic, long offset, int queueId, String brokerName) {
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java
index 1ae0f7aed..63b188e64 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java
@@ -212,16 +212,16 @@ public class TransactionalMessageServiceImpl implements TransactionalMessageServ
}
}
} else {
- if ((0 <= valueOfCurrentMinusBorn) && (valueOfCurrentMinusBorn < checkImmunityTime)) {
+ if (0 <= valueOfCurrentMinusBorn && valueOfCurrentMinusBorn < checkImmunityTime) {
log.debug("New arrived, the miss offset={}, check it later checkImmunity={}, born={}", i,
checkImmunityTime, new Date(msgExt.getBornTimestamp()));
break;
}
}
List<MessageExt> opMsg = pullResult.getMsgFoundList();
- boolean isNeedCheck = (opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime)
- || (opMsg != null && (opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout))
- || (valueOfCurrentMinusBorn <= -1);
+ boolean isNeedCheck = opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime
+ || opMsg != null && opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout
+ || valueOfCurrentMinusBorn <= -1;
if (isNeedCheck) {
if (!putBackHalfMsgQueue(msgExt, i)) {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
index df2b2a6b7..4f4f98c98 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
@@ -438,7 +438,7 @@ public class ConsumeMessageOrderlyService implements ConsumeMessageService {
final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
synchronized (objLock) {
if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
- || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
+ || this.processQueue.isLocked() && !this.processQueue.isLockExpired()) {
final long beginTime = System.currentTimeMillis();
for (boolean continueConsume = true; continueConsume; ) {
if (this.processQueue.isDropped()) {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
index f1a8e866c..789852a6c 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
@@ -726,8 +726,8 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
private void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName, final MessageQueue mq)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
try {
- if ((brokerName != null && brokerName.startsWith(MixAll.LOGICAL_QUEUE_MOCK_BROKER_PREFIX))
- || (mq != null && mq.getBrokerName().startsWith(MixAll.LOGICAL_QUEUE_MOCK_BROKER_PREFIX))) {
+ if (brokerName != null && brokerName.startsWith(MixAll.LOGICAL_QUEUE_MOCK_BROKER_PREFIX)
+ || mq != null && mq.getBrokerName().startsWith(MixAll.LOGICAL_QUEUE_MOCK_BROKER_PREFIX)) {
sendMessageBackAsNormalMessage(msg);
} else {
String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName)
diff --git a/common/src/main/java/org/apache/rocketmq/common/DataVersion.java b/common/src/main/java/org/apache/rocketmq/common/DataVersion.java
index 6d53bd821..73f409e7d 100644
--- a/common/src/main/java/org/apache/rocketmq/common/DataVersion.java
+++ b/common/src/main/java/org/apache/rocketmq/common/DataVersion.java
@@ -82,7 +82,7 @@ public class DataVersion extends RemotingSerializable {
return counter.longValue() == version.counter.longValue();
}
- return (null == counter) && (null == version.counter);
+ return null == counter && null == version.counter;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/filter/impl/PolishExpr.java b/common/src/main/java/org/apache/rocketmq/common/filter/impl/PolishExpr.java
index cdc6187a1..388d095de 100644
--- a/common/src/main/java/org/apache/rocketmq/common/filter/impl/PolishExpr.java
+++ b/common/src/main/java/org/apache/rocketmq/common/filter/impl/PolishExpr.java
@@ -97,8 +97,8 @@ public class PolishExpr {
for (int i = 0; i < size; i++) {
int chValue = (int) expression.charAt(i);
- if ((97 <= chValue && chValue <= 122) || (65 <= chValue && chValue <= 90)
- || (49 <= chValue && chValue <= 57) || 95 == chValue) {
+ if (97 <= chValue && chValue <= 122 || 65 <= chValue && chValue <= 90
+ || 49 <= chValue && chValue <= 57 || 95 == chValue) {
if (Type.OPERATOR == preType || Type.SEPAERATOR == preType || Type.NULL == preType
|| Type.PARENTHESIS == preType) {
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/body/ConsumerRunningInfo.java b/common/src/main/java/org/apache/rocketmq/common/protocol/body/ConsumerRunningInfo.java
index 4a68b25a5..79206727d 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/body/ConsumerRunningInfo.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/body/ConsumerRunningInfo.java
@@ -141,7 +141,7 @@ public class ConsumerRunningInfo extends RemotingSerializable {
mq,
System.currentTimeMillis() - pq.getLastLockTimestamp()));
} else {
- if (pq.isDroped() && (pq.getTryUnlockTimes() > 0)) {
+ if (pq.isDroped() && pq.getTryUnlockTimes() > 0) {
sb.append(String.format("%s %s unlock %d times, still failed%n",
clientId,
mq,
diff --git a/common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingDetail.java b/common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingDetail.java
index f0f27a5cd..40e50f4b4 100644
--- a/common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingDetail.java
+++ b/common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingDetail.java
@@ -97,8 +97,8 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
public static boolean checkIfAsPhysical(TopicQueueMappingDetail mappingDetail, Integer globalId) {
List<LogicQueueMappingItem> mappingItems = getMappingInfo(mappingDetail, globalId);
return mappingItems == null
- || (mappingItems.size() == 1
- && mappingItems.get(0).getLogicOffset() == 0);
+ || mappingItems.size() == 1
+ && mappingItems.get(0).getLogicOffset() == 0;
}
public ConcurrentMap<Integer, List<LogicQueueMappingItem>> getHostedQueues() {
diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManager.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManager.java
index c369d8353..c525c2c36 100644
--- a/controller/src/main/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManager.java
+++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManager.java
@@ -140,7 +140,7 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
long realConfirmOffset = Optional.ofNullable(confirmOffset).orElse(-1L);
prev.setLastUpdateTimestamp(System.currentTimeMillis());
- if (realEpoch > prev.getEpoch() || (realEpoch == prev.getEpoch() && realMaxOffset > prev.getMaxOffset())) {
+ if (realEpoch > prev.getEpoch() || realEpoch == prev.getEpoch() && realMaxOffset > prev.getMaxOffset()) {
prev.setEpoch(realEpoch);
prev.setMaxOffset(realMaxOffset);
prev.setConfirmOffset(realConfirmOffset);
diff --git a/logging/src/main/java/org/apache/rocketmq/logging/inner/LoggingBuilder.java b/logging/src/main/java/org/apache/rocketmq/logging/inner/LoggingBuilder.java
index 996551e77..50b72119c 100644
--- a/logging/src/main/java/org/apache/rocketmq/logging/inner/LoggingBuilder.java
+++ b/logging/src/main/java/org/apache/rocketmq/logging/inner/LoggingBuilder.java
@@ -175,7 +175,7 @@ public class LoggingBuilder {
}
public void append(final LoggingEvent event) {
- if ((dispatcher == null) || !dispatcher.isAlive() || (bufferSize <= 0)) {
+ if (dispatcher == null || !dispatcher.isAlive() || bufferSize <= 0) {
synchronized (appenderPipeline) {
appenderPipeline.appendLoopOnAppenders(event);
}
@@ -383,7 +383,7 @@ public class LoggingBuilder {
int bufferSize = buffer.size();
isActive = !parent.closed;
- while ((bufferSize == 0) && isActive) {
+ while (bufferSize == 0 && isActive) {
buffer.wait();
bufferSize = buffer.size();
isActive = !parent.closed;
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ClientRequestProcessor.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ClientRequestProcessor.java
index 007d49a75..3642d5f59 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ClientRequestProcessor.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/ClientRequestProcessor.java
@@ -61,7 +61,7 @@ public class ClientRequestProcessor implements NettyRequestProcessor {
byte[] content;
Boolean standardJsonOnly = requestHeader.getAcceptStandardJsonOnly();
- if (request.getVersion() >= MQVersion.Version.V4_9_4.ordinal() || (null != standardJsonOnly && standardJsonOnly)) {
+ if (request.getVersion() >= MQVersion.Version.V4_9_4.ordinal() || null != standardJsonOnly && standardJsonOnly) {
content = topicRouteData.encode(SerializerFeature.BrowserCompatible,
SerializerFeature.QuoteFieldNames, SerializerFeature.SkipTransientField,
SerializerFeature.MapSortField);
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
index c1b59993f..feee13f1a 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
@@ -157,7 +157,7 @@ public class DefaultPromise<V> implements Promise<V> {
}
private boolean isSuccess() {
- return isDone() && (exception == null);
+ return isDone() && exception == null;
}
private void timeoutSoCancel() {
diff --git a/pom.xml b/pom.xml
index feeb8d3ed..8d19e968a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -147,7 +147,7 @@
<maven-compiler-plugin.version>3.5.1</maven-compiler-plugin.version>
<maven-source-plugin.version>3.0.1</maven-source-plugin.version>
<maven-help-plugin.version>2.2</maven-help-plugin.version>
- <maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
+ <maven-checkstyle-plugin.version>3.2.0</maven-checkstyle-plugin.version>
<apache-rat-plugin.version>0.12</apache-rat-plugin.version>
<maven-resources-plugin.version>3.0.2</maven-resources-plugin.version>
<coveralls-maven-plugin.version>4.3.0</coveralls-maven-plugin.version>
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAService.java
index 8408b7608..fcac067c5 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAService.java
@@ -99,8 +99,8 @@ public class DefaultHAService implements HAService {
boolean result = this.connectionCount.get() > 0;
result =
result
- && ((masterPutWhere - this.push2SlaveMaxOffset.get()) < this.defaultMessageStore
- .getMessageStoreConfig().getHaMaxGapNotInSync());
+ && masterPutWhere - this.push2SlaveMaxOffset.get() < this.defaultMessageStore
+ .getMessageStoreConfig().getHaMaxGapNotInSync();
return result;
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java
index 1684eed12..ac675a5bf 100644
--- a/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java
+++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java
@@ -180,8 +180,8 @@ public class IndexFile {
public boolean isTimeMatched(final long begin, final long end) {
boolean result = begin < this.indexHeader.getBeginTimestamp() && end > this.indexHeader.getEndTimestamp();
- result = result || (begin >= this.indexHeader.getBeginTimestamp() && begin <= this.indexHeader.getEndTimestamp());
- result = result || (end >= this.indexHeader.getBeginTimestamp() && end <= this.indexHeader.getEndTimestamp());
+ result = result || begin >= this.indexHeader.getBeginTimestamp() && begin <= this.indexHeader.getEndTimestamp();
+ result = result || end >= this.indexHeader.getBeginTimestamp() && end <= this.indexHeader.getEndTimestamp();
return result;
}
@@ -219,7 +219,7 @@ public class IndexFile {
timeDiff *= 1000L;
long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
- boolean timeMatched = (timeRead >= begin) && (timeRead <= end);
+ boolean timeMatched = timeRead >= begin && timeRead <= end;
if (keyHash == keyHashRead && timeMatched) {
phyOffsets.add(phyOffsetRead);
diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java
index fa8d680fd..4cd0088b1 100644
--- a/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java
@@ -504,7 +504,7 @@ public class BatchConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCy
}
long behind = System.currentTimeMillis() - storeTime;
- if (behind > 10000 && (System.currentTimeMillis() % 10000 == 0)) {
+ if (behind > 10000 && System.currentTimeMillis() % 10000 == 0) {
String flag = "LEVEL" + (behind / 10000);
log.warn("Reput behind {} topic:{} queue:{} offset:{} behind:{}", flag, topic, queueId, offset, behind);
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
index 4b14c8f3c..aa1fa37b5 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -1637,7 +1637,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
Iterator<Entry<String, TopicConfig>> iterator = topicConfigSerializeWrapper.getTopicConfigTable().entrySet().iterator();
while (iterator.hasNext()) {
String topic = iterator.next().getKey();
- if (topicList.getTopicList().contains(topic) || (!specialTopic && (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)))) {
+ if (topicList.getTopicList().contains(topic) || !specialTopic && (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX))) {
iterator.remove();
}
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
index ae49ae35e..0b74a3e1c 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
@@ -294,7 +294,7 @@ public class MQAdminStartup {
private static SubCommand findSubCommand(final String name) {
for (SubCommand cmd : SUB_COMMANDS) {
- if (cmd.commandName().equalsIgnoreCase(name) || (cmd.commandAlias() != null && cmd.commandAlias().equalsIgnoreCase(name))) {
+ if (cmd.commandName().equalsIgnoreCase(name) || cmd.commandAlias() != null && cmd.commandAlias().equalsIgnoreCase(name)) {
return cmd;
}
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/stats/StatsAllSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/stats/StatsAllSubCommand.java
index 2e1cdaec2..1fcf791aa 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/stats/StatsAllSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/stats/StatsAllSubCommand.java
@@ -91,8 +91,8 @@ public class StatsAllSubCommand implements SubCommand {
} catch (Exception e) {
}
- if (!activeTopic || (inMsgCntToday > 0) ||
- (outMsgCntToday > 0)) {
+ if (!activeTopic || inMsgCntToday > 0 ||
+ outMsgCntToday > 0) {
System.out.printf("%-64s %-64s %12d %11.2f %11.2f %14d %14d%n",
UtilAll.frontStringAtLeast(topic, 64),
@@ -106,7 +106,7 @@ public class StatsAllSubCommand implements SubCommand {
}
}
} else {
- if (!activeTopic || (inMsgCntToday > 0)) {
+ if (!activeTopic || inMsgCntToday > 0) {
System.out.printf("%-64s %-64s %12d %11.2f %11s %14d %14s%n",
UtilAll.frontStringAtLeast(topic, 64),
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
index 639191add..c0d0fbe46 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
@@ -141,7 +141,7 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
try {
defaultMQAdminExt.start();
- if ((!commandLine.hasOption("b") && !commandLine.hasOption('c'))
+ if (!commandLine.hasOption("b") && !commandLine.hasOption('c')
|| !commandLine.hasOption("qn")) {
ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
return;
[rocketmq] 07/07: [ISSUE #5104] Clean property PROPERTY_TIMER_DELAY_MS if not timer message #
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit a174e0bfc27fbfb1b67b393c7fff39e65eb46abc
Author: Oliver <wq...@163.com>
AuthorDate: Mon Sep 19 10:01:09 2022 +0800
[ISSUE #5104] Clean property PROPERTY_TIMER_DELAY_MS if not timer message #
---
broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java b/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java
index 78a1ee2cd..f8a5f6789 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/util/HookUtils.java
@@ -155,6 +155,9 @@ public class HookUtils {
if (null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_SEC)) {
MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_TIMER_DELAY_SEC);
}
+ if (null != msg.getProperty(MessageConst.PROPERTY_TIMER_DELAY_MS)) {
+ MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_TIMER_DELAY_MS);
+ }
return false;
//return this.defaultMessageStore.getMessageStoreConfig().isTimerInterceptDelayLevel();
}
[rocketmq] 04/07: Sync commons-lang3 version (#5103)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit 02aa4758bb6b12e1fbf4a55ea262e5aa7a2067b5
Author: Zhanhui Li <li...@gmail.com>
AuthorDate: Sat Sep 17 15:12:23 2022 +0800
Sync commons-lang3 version (#5103)
---
WORKSPACE | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/WORKSPACE b/WORKSPACE
index 231336eae..a7c3fa125 100644
--- a/WORKSPACE
+++ b/WORKSPACE
@@ -49,7 +49,7 @@ maven_install(
"com.github.luben:zstd-jni:1.5.2-2",
"org.lz4:lz4-java:1.8.0",
"commons-validator:commons-validator:1.7",
- "org.apache.commons:commons-lang3:3.4",
+ "org.apache.commons:commons-lang3:3.12.0",
"org.hamcrest:hamcrest-core:1.3",
"io.openmessaging.storage:dledger:0.3.1",
"net.java.dev.jna:jna:4.2.2",
[rocketmq] 06/07: Fixed intermixing of punctuation marks in English documents (#5086)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit 2aa8124bec1ce1f2671d487f722110b76716ad4b
Author: zhaows <41...@users.noreply.github.com>
AuthorDate: Sun Sep 18 11:23:48 2022 +0800
Fixed intermixing of punctuation marks in English documents (#5086)
---
..._Topic_Logic_Queue_\350\256\276\350\256\241.md" | 2 +-
docs/en/Configuration_TLS.md | 8 ++++----
docs/en/Design_Remoting.md | 2 +-
docs/en/Design_Store.md | 2 +-
docs/en/FAQ.md | 2 +-
docs/en/Operations_Trace.md | 6 +++---
docs/en/README.md | 22 +++++++++++-----------
docs/en/Troubleshoopting.md | 4 ++--
docs/en/architecture.md | 18 +++++++++---------
docs/en/best_practice.md | 2 +-
.../client/java/API_Reference_DefaultMQProducer.md | 2 +-
docs/en/dledger/deploy_guide.md | 4 ++--
docs/en/dledger/quick_start.md | 6 +++---
docs/en/operation.md | 18 +++++++++---------
14 files changed, 49 insertions(+), 49 deletions(-)
diff --git "a/docs/cn/statictopic/RocketMQ_Static_Topic_Logic_Queue_\350\256\276\350\256\241.md" "b/docs/cn/statictopic/RocketMQ_Static_Topic_Logic_Queue_\350\256\276\350\256\241.md"
index 67cb1d587..784a83f92 100644
--- "a/docs/cn/statictopic/RocketMQ_Static_Topic_Logic_Queue_\350\256\276\350\256\241.md"
+++ "b/docs/cn/statictopic/RocketMQ_Static_Topic_Logic_Queue_\350\256\276\350\256\241.md"
@@ -2,7 +2,7 @@
| 时间 | 主要内容 | 作者 |
| --- | --- | --- |
| 2021-11-01 | 初稿，包括背景、目标、SOT定义与持久化、SOT生命周期、SOT的使用、API逻辑修改、问题与风险 | dongeforever |
-| 2021-11-15 | 修改 LogicQueue 的定义，不要引入新字段，完全复用旧的MessageQueue; RemappingStaticTopic时，不要迁移『位点』『幂等数据』等，而是采用Double-Check-Read 的机制| dongforever |
+| 2021-11-15 | 修改 LogicQueue 的定义，不要引入新字段，完全复用旧的MessageQueue；RemappingStaticTopic时，不要迁移『位点』『幂等数据』等，而是采用Double-Check-Read 的机制| dongforever |
| 2021-12-01 | 更新问题与风险，增加关于一致性、OutOfRange、拉取中断的详细说明| dongforever |
| 2021-12-03 | 增加代码走读的说明| dongforever |
| 2021-12-10 | 引入Scope概念，保留『多集群动态零耦合』的集群设计模型 | dongforever |
diff --git a/docs/en/Configuration_TLS.md b/docs/en/Configuration_TLS.md
index 67b5b9e1b..445d186d2 100644
--- a/docs/en/Configuration_TLS.md
+++ b/docs/en/Configuration_TLS.md
@@ -34,7 +34,7 @@ openssl pkcs8 -topk8 -v1 PBE-SHA1-RC4-128 -in client_rsa.key -out client.key
```
## 2 Create tls.properties
-Create tls.properties，correctly configure the path and password of the generated certificates.
+Create tls.properties, correctly configure the path and password of the generated certificates.
```properties
# The flag to determine whether use test mode when initialize TLS context. default is true
@@ -76,13 +76,13 @@ Edit the configuration file under the rocketmq/bin path to make tls.properties c
The value of "tls.config.file" needs to be replaced by the file path created in step 2.
### 3.1 Edit runserver.sh
-Add following content in JAVA_OPT：
+Add following content in JAVA_OPT:
```shell
JAVA_OPT="${JAVA_OPT} -Dtls.server.mode=enforcing -Dtls.config.file=/opt/rocketmq-4.9.3/conf/tls.properties"
```
### 3.2 Edit runbroker.sh
-Add following content in JAVA_OPT：
+Add following content in JAVA_OPT:
```shell
JAVA_OPT="${JAVA_OPT} -Dorg.apache.rocketmq.remoting.ssl.mode=enforcing -Dtls.config.file=/opt/rocketmq-4.9.3/conf/tls.properties -Dtls.enable=true"
@@ -102,7 +102,7 @@ tls.client.certPath=/opt/certFiles/client.pem
tls.client.trustCertPath=/opt/certFiles/ca.pem
```
-Add following parameters in JVM. The value of "tls.config.file" needs to be replaced by the file path we created：
+Add following parameters in JVM. The value of "tls.config.file" needs to be replaced by the file path we created:
```properties
-Dtls.client.authServer=true -Dtls.enable=true -Dtls.test.mode.enable=false -Dtls.config.file=/opt/certs/tlsclient.properties
```
diff --git a/docs/en/Design_Remoting.md b/docs/en/Design_Remoting.md
index b7b8c86cb..46e001a1d 100644
--- a/docs/en/Design_Remoting.md
+++ b/docs/en/Design_Remoting.md
@@ -28,7 +28,7 @@ From the above figure, the transport content can be divided into four parts:
(1) Message length: total length, four bytes of storage, occupying an int type;
-(2) Serialization type header length: occupying an int type. The first byte represents the serialization type, and the last three bytes represent the header length；
+(2) Serialization type header length: occupying an int type. The first byte represents the serialization type, and the last three bytes represent the header length;
(3) Header data: serialized header data;
diff --git a/docs/en/Design_Store.md b/docs/en/Design_Store.md
index cbc661305..747c98e6f 100644
--- a/docs/en/Design_Store.md
+++ b/docs/en/Design_Store.md
@@ -14,7 +14,7 @@ Message storage is the most complicated and important part of RocketMQ. This sec
The message storage architecture diagram consists of 3 files related to message storage: `CommitLog` file, `ConsumeQueue` file, and `IndexFile`.
-* `CommitLog`：The `CommitLog` file stores message body and metadata sent by producer, and the message content is not fixed length. The default size of one `CommitLog` file is 1G, the length of the file name is 20 digits, the left side is zero padded, and the remaining is the starting offset. For example, `00000000000000000000` represents the first file, the starting offset is 0, and the file size is 1G=1073741824, when the first `CommitLog` file is full, the second `CommitLog` file is `0 [...]
+* `CommitLog`: The `CommitLog` file stores message body and metadata sent by producer, and the message content is not fixed length. The default size of one `CommitLog` file is 1G, the length of the file name is 20 digits, the left side is zero padded, and the remaining is the starting offset. For example, `00000000000000000000` represents the first file, the starting offset is 0, and the file size is 1G=1073741824, when the first `CommitLog` file is full, the second `CommitLog` file is ` [...]
* `ConsumeQueue`: The `ConsumeQueue` is used to improve the performance of message consumption. Since RocketMQ uses topic-based subscription mode, message consumption is specific to the topic. Traversing the commitlog file to retrieve messages of one topic is very inefficient. The consumer can find the messages to be consumed according to the `ConsumeQueue`. The `ConsumeQueue`(logic consume queue) as an index of the consuming message stores the starting physical offset `offset` in `Commi [...]
* `IndexFile`: The `IndexFile` provides a way to query messages by key or time interval. The path of the `IndexFile` is `$HOME/store/index/${fileName}`, the file name `fileName` is named after the timestamp when it was created. One IndexFile's size is about 400M, and it can store 2000W indexes. The underlying storage of `IndexFile` is designed to implement the `HashMap` structure in the file system, so RocketMQ's index file is a hash index.
diff --git a/docs/en/FAQ.md b/docs/en/FAQ.md
index 97b9a0e05..dac53ecbf 100644
--- a/docs/en/FAQ.md
+++ b/docs/en/FAQ.md
@@ -70,7 +70,7 @@ consumer.setConsumeThreadMax(20);
### 1. If you start a producer or consumer failed and the error message is producer group or consumer repeat.
-Reason：Using the same Producer /Consumer Group to launch multiple instances of Producer/Consumer in the same JVM may cause the client fail to start.
+Reason: Using the same Producer /Consumer Group to launch multiple instances of Producer/Consumer in the same JVM may cause the client fail to start.
Solution: Make sure that a JVM corresponding to one Producer /Consumer Group starts only with one Producer/Consumer instance.
diff --git a/docs/en/Operations_Trace.md b/docs/en/Operations_Trace.md
index 5cc003ce4..74c2cde1a 100644
--- a/docs/en/Operations_Trace.md
+++ b/docs/en/Operations_Trace.md
@@ -46,10 +46,10 @@ For scenarios with large amount of trace message data , one of the Broker nodes
RocketMQ's message trace feature supports two ways to store trace data:
### 3.1 System-level TraceTopic
-By default, message track data is stored in the system-level TraceTopic(names：**RMQ_SYS_TRACE_TOPIC**)。This Topic is automatically created when the Broker node is started(As described above, the switch variable **traceTopicEnable** needs to be set to **true** in the Broker configuration file）。
+By default, message track data is stored in the system-level TraceTopic(names: **RMQ_SYS_TRACE_TOPIC**).This Topic is automatically created when the Broker node is started(As described above, the switch variable **traceTopicEnable** needs to be set to **true** in the Broker configuration file）.
### 3.2 Custom TraceTopic
-If the user is not prepared to store the message track data in the system-level default TraceTopic, you can also define and create a user-level Topic to save the track (that is, to create a regular Topic to save the message track data)。The following section introduces how the Client interface supports the user-defined TraceTopic.
+If the user is not prepared to store the message track data in the system-level default TraceTopic, you can also define and create a user-level Topic to save the track (that is, to create a regular Topic to save the message track data).The following section introduces how the Client interface supports the user-defined TraceTopic.
## 4 Client Practices that Support Message Trace
In order to reduce as much as possible the transformation work of RocketMQ message trace feature used in the user service system, the author added a switch parameter (**enableMsgTrace**) to the original interface in the design to realize whether the message trace is opened or not.
@@ -95,7 +95,7 @@ In order to reduce as much as possible the transformation work of RocketMQ messa
The initialization of `DefaultMQProducer` and `DefaultMQPushConsumer` instances can be changed to support the custom storage message trace Topic as follows when sending and subscribing messages above.
```
- ##Where Topic_test11111 needs to be pre-created by the user to save the message trace；
+ ##Where Topic_test11111 needs to be pre-created by the user to save the message trace;
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true,"Topic_test11111");
......
diff --git a/docs/en/README.md b/docs/en/README.md
index e1e569e5a..2a096c334 100644
--- a/docs/en/README.md
+++ b/docs/en/README.md
@@ -5,39 +5,39 @@
### 1. Concepts & Features
-- [Concept](Concept.md)：introduce basic concepts in RocketMQ.
+- [Concept](Concept.md): introduce basic concepts in RocketMQ.
-- [Feature](Feature.md)：introduce functional features of RocketMQ's implementations.
+- [Feature](Feature.md): introduce functional features of RocketMQ's implementations.
### 2. Architecture Design
-- [Architecture](architecture.md)：introduce RocketMQ's deployment and technical architecture.
+- [Architecture](architecture.md): introduce RocketMQ's deployment and technical architecture.
-- [Design](design.md)：introduce design concept of RocketMQ's key mechanisms, including message storage, communication mechanisms, message filter, loadbalance, transaction message, etc.
+- [Design](design.md): introduce design concept of RocketMQ's key mechanisms, including message storage, communication mechanisms, message filter, loadbalance, transaction message, etc.
### 3. Example
-- [Example](RocketMQ_Example.md) ：introduce RocketMQ's common usage, including basic example, sequence message example, delay message example, batch message example, filter message example, transaction message example, etc.
+- [Example](RocketMQ_Example.md): introduce RocketMQ's common usage, including basic example, sequence message example, delay message example, batch message example, filter message example, transaction message example, etc.
### 4. Best Practice
-- [Best Practice](best_practice.md)：introduce RocketMQ's best practice, including producer, consumer, broker, NameServer, configuration of client, and the best parameter configuration of JVM, linux.
+- [Best Practice](best_practice.md): introduce RocketMQ's best practice, including producer, consumer, broker, NameServer, configuration of client, and the best parameter configuration of JVM, linux.
-- [Message Trace](msg_trace/user_guide.md)：introduce how to use RocketMQ's message tracing feature.
+- [Message Trace](msg_trace/user_guide.md): introduce how to use RocketMQ's message tracing feature.
-- [Auth Management](acl/Operations_ACL.md)：introduce how to deployment quickly and how to use RocketMQ cluster enabling auth management feature.
+- [Auth Management](acl/Operations_ACL.md): introduce how to deployment quickly and how to use RocketMQ cluster enabling auth management feature.
-- [Quick Start](dledger/quick_start.md)：introduce how to deploy Dledger quickly.
+- [Quick Start](dledger/quick_start.md): introduce how to deploy Dledger quickly.
-- [Cluster Deployment](dledger/deploy_guide.md)：introduce how to deploy Dledger in cluster.
+- [Cluster Deployment](dledger/deploy_guide.md): introduce how to deploy Dledger in cluster.
- [Proxy Deployment](proxy/deploy_guide.md)
Introduce how to deploy proxy (both `Local` mode and `Cluster` mode).
### 5. Operation and maintenance management
-- [Operation](operation.md)：introduce RocketMQ's deployment modes that including single-master mode, multi-master mode, multi-master multi-slave mode and so on, as well as the usage of operation tool mqadmin.
+- [Operation](operation.md): introduce RocketMQ's deployment modes that including single-master mode, multi-master mode, multi-master multi-slave mode and so on, as well as the usage of operation tool mqadmin.
### 6. API Reference（TODO）
diff --git a/docs/en/Troubleshoopting.md b/docs/en/Troubleshoopting.md
index 085b8288a..ee4adab8c 100644
--- a/docs/en/Troubleshoopting.md
+++ b/docs/en/Troubleshoopting.md
@@ -2,7 +2,7 @@
## 1 RocketMQ's mqadmin command error.
-> Problem: Sometimes after deploying the RocketMQ cluster, when you try to execute some commands of "mqadmin", the following exception will appear：
+> Problem: Sometimes after deploying the RocketMQ cluster, when you try to execute some commands of "mqadmin", the following exception will appear:
>
> ```java
> org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <null> failed
@@ -72,5 +72,5 @@ After sending message with RocketMQ, you will usually see the following log:
SendResult [sendStatus=SEND_OK, msgId=0A42333A0DC818B4AAC246C290FD0000, offsetMsgId=0A42333A00002A9F000000000134F1F5, messageQueue=MessageQueue [topic=topicTest1, BrokerName=mac.local, queueId=3], queueOffset=4]
```
-- msgId，for the client, the msgId is generated by the producer instance. Specifically, the method `MessageClientIDSetter.createUniqIDBuffer()` is called to generate a unique Id.
+- msgId, for the client, the msgId is generated by the producer instance. Specifically, the method `MessageClientIDSetter.createUniqIDBuffer()` is called to generate a unique Id.
- offsetMsgId, offsetMsgId is generated by the Broker server when writing a message ( string consists of "IP address + port" and "CommitLog's physical offset address"), and offsetMsgId is the messageId used to query in the RocketMQ console.
diff --git a/docs/en/architecture.md b/docs/en/architecture.md
index 18e1cc002..863a62200 100644
--- a/docs/en/architecture.md
+++ b/docs/en/architecture.md
@@ -6,18 +6,18 @@
The RocketMQ architecture is divided into four parts, as shown in the figure above:
-- Producer：The role of message publishing supports distributed cluster mode deployment. Producer selects the corresponding Broker cluster queue for message delivery through MQ's load balancing module. The delivery process supports fast failure and low latency.
+- Producer: The role of message publishing supports distributed cluster mode deployment. Producer selects the corresponding Broker cluster queue for message delivery through MQ's load balancing module. The delivery process supports fast failure and low latency.
-- Consumer：The role of message consumption supports distributed cluster deployment. Support push, pull two modes to consume messages. It also supports cluster mode and broadcast mode consumption, and it provides a real-time message subscription mechanism to meet the needs of most users.
+- Consumer: The role of message consumption supports distributed cluster deployment. Support push, pull two modes to consume messages. It also supports cluster mode and broadcast mode consumption, and it provides a real-time message subscription mechanism to meet the needs of most users.
-- NameServer：NameServer is a very simple Topic routing registry with a role similar to ZooKeeper in Dubbo, which supports dynamic registration and discovery of Broker. It mainly includes two functions: Broker management, NameServer accepts the registration information of the Broker cluster and saves it as the basic data of the routing information. Then provide a heartbeat detection mechanism to check whether the broker is still alive; routing information management, each NameServer will [...]
+- NameServer: NameServer is a very simple Topic routing registry with a role similar to ZooKeeper in Dubbo, which supports dynamic registration and discovery of Broker. It mainly includes two functions: Broker management, NameServer accepts the registration information of the Broker cluster and saves it as the basic data of the routing information. Then provide a heartbeat detection mechanism to check whether the broker is still alive; routing information management, each NameServer will [...]
-- BrokerServer：Broker is responsible for the storage, delivery and query of messages and high availability guarantees. In order to achieve these functions, Broker includes the following important sub-modules.
-1. Remoting Module：The entire broker entity handles requests from the clients side.
-2. Client Manager：Topic subscription information for managing the client (Producer/Consumer) and maintaining the Consumer
-3. Store Service：Provides a convenient and simple API interface for handling message storage to physical hard disks and query functions.
-4. HA Service：Highly available service that provides data synchronization between Master Broker and Slave Broker.
-5. Index Service：The message delivered to the Broker is indexed according to a specific Message key to provide a quick query of the message.
+- BrokerServer: Broker is responsible for the storage, delivery and query of messages and high availability guarantees. In order to achieve these functions, Broker includes the following important sub-modules.
+1. Remoting Module: The entire broker entity handles requests from the clients side.
+2. Client Manager: Topic subscription information for managing the client (Producer/Consumer) and maintaining the Consumer
+3. Store Service: Provides a convenient and simple API interface for handling message storage to physical hard disks and query functions.
+4. HA Service: Highly available service that provides data synchronization between Master Broker and Slave Broker.
+5. Index Service: The message delivered to the Broker is indexed according to a specific Message key to provide a quick query of the message.
diff --git a/docs/en/best_practice.md b/docs/en/best_practice.md
index 5d70aeed3..da279e297 100755
--- a/docs/en/best_practice.md
+++ b/docs/en/best_practice.md
@@ -72,7 +72,7 @@ Thirdly, the producer is a virtual machine with low reliability, which is not su
In conclusion, it is recommended that the retry process must be controlled by the application.
### 1.3 Send message by oneway
-Typically, this is the process by which messages are sent：
+Typically, this is the process by which messages are sent:
- Client send request to server
- Server process request
diff --git a/docs/en/client/java/API_Reference_DefaultMQProducer.md b/docs/en/client/java/API_Reference_DefaultMQProducer.md
index 0feaff24c..edc141df8 100644
--- a/docs/en/client/java/API_Reference_DefaultMQProducer.md
+++ b/docs/en/client/java/API_Reference_DefaultMQProducer.md
@@ -6,7 +6,7 @@
extends ClientConfig
implements MQProducer`
->`DefaultMQProducer` is the entry point for an application to post messages, out of the box，ca quickly create a producer with a no-argument construction. it is mainly responsible for message sending, support synchronous、asynchronous、one-way send. All of these send methods support batch send. The parameters of the sender can be adjusted through the getter/setter methods , provided by this class. `DefaultMQProducer` has multi send method and each method is slightly different. Make sure [...]
+>`DefaultMQProducer` is the entry point for an application to post messages, out of the box, ca quickly create a producer with a no-argument construction. it is mainly responsible for message sending, support synchronous、asynchronous、one-way send. All of these send methods support batch send. The parameters of the sender can be adjusted through the getter/setter methods , provided by this class. `DefaultMQProducer` has multi send method and each method is slightly different. Make sure [...]
``` java
public class Producer {
diff --git a/docs/en/dledger/deploy_guide.md b/docs/en/dledger/deploy_guide.md
index 55509498d..06cf333a6 100644
--- a/docs/en/dledger/deploy_guide.md
+++ b/docs/en/dledger/deploy_guide.md
@@ -1,7 +1,7 @@
# Dledger cluster deployment
---
## preface
-This document introduces how to deploy auto failover RocketMQ-on-DLedger Group。
+This document introduces how to deploy auto failover RocketMQ-on-DLedger Group.
RocketMQ-on-DLedger Group is a broker group with **same name**, needs at least 3 nodes, elect a Leader by Raft algorithm automatically, the others as Follower, replicating data between Leader and Follower for system high available.
RocketMQ-on-DLedger Group can failover automatically, and maintains consistent.
@@ -12,7 +12,7 @@ RocketMQ-on-DLedger Group can scale up horizontal, that is, can deploy any Rocke
#### 1.1 Write the configuration
each RocketMQ-on-DLedger Group needs at least 3 machines.(assuming 3 in this document)
write 3 configuration files, advising refer to the directory of conf/dledger 's example configuration file.
-key configuration items：
+key configuration items:
| name | meaning | example |
| --- | --- | --- |
diff --git a/docs/en/dledger/quick_start.md b/docs/en/dledger/quick_start.md
index 83fa76236..dfc7894af 100644
--- a/docs/en/dledger/quick_start.md
+++ b/docs/en/dledger/quick_start.md
@@ -3,7 +3,7 @@
### preface
This document is mainly introduced for how to build and deploy auto failover RocketMQ cluster based on DLedger.
-For detailed new cluster deployment and old cluster upgrade document, please refer to [Deployment Guide](deploy_guide.md)。
+For detailed new cluster deployment and old cluster upgrade document, please refer to [Deployment Guide](deploy_guide.md).
### 1. Build from source code
Build phase contains two parts, first, build DLedger, then build RocketMQ.
@@ -45,7 +45,7 @@ If everything goes well, the following content will appear:
-（BID is 0 indicate Master，the others are Follower）
+（BID is 0 indicate Master, the others are Follower）
After startup successful, producer can produce message, and then test failover scenario.
@@ -55,7 +55,7 @@ Stop cluster fastly, execute the following command:
$ sh bin/dledger/fast-try.sh stop
```
-Quick deployment, default configuration is in directory conf/dledger，default storage path is /tmp/rmqstore.
+Quick deployment, default configuration is in directory conf/dledger, default storage path is /tmp/rmqstore.
### 3. Failover
diff --git a/docs/en/operation.md b/docs/en/operation.md
index 9dff41ab7..799862376 100644
--- a/docs/en/operation.md
+++ b/docs/en/operation.md
@@ -34,9 +34,9 @@ The broker[broker-a, 192.169.1.2:10911] boot success...
Cluster contains Master node only, no Slave node, eg: 2 Master nodes, 3 Master nodes, advantages and disadvantages of this mode are shown below:
-- advantages：simple configuration, single Master node broke down or restart do not impact application. Under RAID10 disk config, even if machine broken down and cannot recover, message do not get lost because of RAID10's high reliable(async flush to disk lost little message, sync to disk do not lost message), this mode get highest performance.
+- advantages: simple configuration, single Master node broke down or restart do not impact application. Under RAID10 disk config, even if machine broken down and cannot recover, message do not get lost because of RAID10's high reliable(async flush to disk lost little message, sync to disk do not lost message), this mode get highest performance.
-- disadvantages：during the machine's down time, messages have not be consumed on this machine can not be subscribed before recovery. That will impacts message's instantaneity.
+- disadvantages: during the machine's down time, messages have not be consumed on this machine can not be subscribed before recovery. That will impacts message's instantaneity.
##### 1）Start NameServer
@@ -54,10 +54,10 @@ The Name Server boot success...
##### 2）start Broker cluster
```bash
-### start the first Master on machine A, eg:NameServer's IP is ：192.168.1.1
+### start the first Master on machine A, eg:NameServer's IP is 192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-a.properties &
-### start the second Master on machine B, eg:NameServer's IP is ：192.168.1.1
+### start the second Master on machine B, eg:NameServer's IP is 192.168.1.1
$ nohup sh mqbroker -n 192.168.1.1:9876 -c $ROCKETMQ_HOME/conf/2m-noslave/broker-b.properties &
...
@@ -139,9 +139,9 @@ The above Broker matches Slave by specifying the same BrokerName, Master's Broke
### 2 mqadmin management tool
-> Attentions：
+> Attentions:
>
-> 1. execute command：`./mqadmin {command} {args}`
+> 1. execute command: `./mqadmin {command} {args}`
> 2. almost all commands need -n indicates NameSerer address, format is ip:port
> 3. almost all commands can get help info by -h
> 4. if command contains both Broker address(-b) and cluster name(-c), it's prior to use broker address. If command do not contains broker address, it will executed on all hosts in this cluster. Support only one broker host. -b format is ip:port, default port is 10911
@@ -414,7 +414,7 @@ The above Broker matches Slave by specifying the same BrokerName, Master's Broke
<td rowspan=8 height=391 class=xl67 width=177 style='border-bottom:1.0pt;
height:292.0pt;border-top:none;width:133pt'>clusterRT</td>
<td rowspan=8 class=xl70 width=175 style='border-bottom:1.0pt;
- border-top:none;width:131pt'>send message to detect each cluster's Broker RT. Message will be sent to ${BrokerName} Topic。</td>
+ border-top:none;width:131pt'>send message to detect each cluster's Broker RT. Message will be sent to ${BrokerName} Topic.</td>
<td class=xl65 width=177 style='width:133pt'>-a</td>
<td class=xl66 width=185 style='width:139pt'>amount, count of detection, RT = sum time /
amount</td>
@@ -1332,7 +1332,7 @@ The above Broker matches Slave by specifying the same BrokerName, Master's Broke
#### 3.1 RocketMQ's mqadmin command error
-> question description：execute mqadmin occur below exception after deploy RocketMQ cluster.
+> question description: execute mqadmin occur below exception after deploy RocketMQ cluster.
>
> ```java
> org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <null> failed
@@ -1345,7 +1345,7 @@ Solution: execute command `export NAMESRV_ADDR=ip:9876` (ip is NameServer's ip a
> question description: one producer produce message, consumer A can consume, consume B cannot consume, RocketMQ console print:
>
> ```java
-> Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message。
+> Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message.
> ```
Solution: make sure that producer and consumer has the same version of rocketmq-client.
[rocketmq] 02/07: [ISSUE #5097] fix illegal reflective access (#5098)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit afea709ac9ab660974d30cc6f10a53b665203e1f
Author: SSpirits <ad...@lv5.moe>
AuthorDate: Fri Sep 16 18:11:22 2022 +0800
[ISSUE #5097] fix illegal reflective access (#5098)
---
.../java/org/apache/rocketmq/common/UtilAll.java | 32 ++++++++++++++--------
pom.xml | 2 +-
2 files changed, 22 insertions(+), 12 deletions(-)
diff --git a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
index 565de7549..8c175e966 100644
--- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
@@ -21,6 +21,7 @@ import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
+import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.net.Inet4Address;
import java.net.Inet6Address;
@@ -42,11 +43,15 @@ import java.util.Map;
import java.util.zip.CRC32;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;
+import org.apache.commons.lang3.JavaVersion;
+import org.apache.commons.lang3.SystemUtils;
import org.apache.commons.validator.routines.InetAddressValidator;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
public class UtilAll {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
@@ -670,7 +675,19 @@ public class UtilAll {
if (buffer == null || !buffer.isDirect() || buffer.capacity() == 0) {
return;
}
- invoke(invoke(viewed(buffer), "cleaner"), "clean");
+ if (SystemUtils.isJavaVersionAtLeast(JavaVersion.JAVA_9)) {
+ try {
+ Field field = Unsafe.class.getDeclaredField("theUnsafe");
+ field.setAccessible(true);
+ Unsafe unsafe = (Unsafe) field.get(null);
+ Method cleaner = method(unsafe, "invokeCleaner", new Class[] {ByteBuffer.class});
+ cleaner.invoke(unsafe, viewed(buffer));
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ } else {
+ invoke(invoke(viewed(buffer), "cleaner"), "clean");
+ }
}
public static Object invoke(final Object target, final String methodName, final Class<?>... args) {
@@ -697,17 +714,10 @@ public class UtilAll {
}
private static ByteBuffer viewed(ByteBuffer buffer) {
- String methodName = "viewedBuffer";
-
- Method[] methods = buffer.getClass().getMethods();
- for (Method method : methods) {
- if (method.getName().equals("attachment")) {
- methodName = "attachment";
- break;
- }
+ if (!buffer.isDirect()) {
+ throw new IllegalArgumentException("buffer is non-direct");
}
-
- ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
+ ByteBuffer viewedBuffer = (ByteBuffer) ((DirectBuffer) buffer).attachment();
if (viewedBuffer == null) {
return buffer;
} else {
diff --git a/pom.xml b/pom.xml
index 6b0e20ca8..feeb8d3ed 100644
--- a/pom.xml
+++ b/pom.xml
@@ -108,7 +108,7 @@
<fastjson.version>1.2.69_noneautotype</fastjson.version>
<javassist.version>3.20.0-GA</javassist.version>
<jna.version>4.2.2</jna.version>
- <commons-lang3.version>3.4</commons-lang3.version>
+ <commons-lang3.version>3.12.0</commons-lang3.version>
<commons-io.version>2.7</commons-io.version>
<guava.version>31.0.1-jre</guava.version>
<openmessaging.version>0.3.1-alpha</openmessaging.version>
[rocketmq] 03/07: Sync DLedger version in Bazel scripts (#5102)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch release-5.0.0
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit ac7e4cbfc9967a9402c204cc26d0d393bb468377
Author: Zhanhui Li <li...@gmail.com>
AuthorDate: Fri Sep 16 21:05:14 2022 +0800
Sync DLedger version in Bazel scripts (#5102)
* Use dledger from maven
* Sync other references
---
WORKSPACE | 11 +++++----
broker/BUILD.bazel | 2 +-
container/BUILD.bazel | 4 ++--
controller/BUILD.bazel | 4 ++--
store/BUILD.bazel | 42 +++++++++++++---------------------
store/libs/dledger-0.2.8-SNAPSHOT.jar | Bin 186014 -> 0 bytes
6 files changed, 27 insertions(+), 36 deletions(-)
diff --git a/WORKSPACE b/WORKSPACE
index 9bc6419b6..231336eae 100644
--- a/WORKSPACE
+++ b/WORKSPACE
@@ -51,7 +51,7 @@ maven_install(
"commons-validator:commons-validator:1.7",
"org.apache.commons:commons-lang3:3.4",
"org.hamcrest:hamcrest-core:1.3",
- # "io.openmessaging.storage:dledger:0.2.4",
+ "io.openmessaging.storage:dledger:0.3.1",
"net.java.dev.jna:jna:4.2.2",
"ch.qos.logback:logback-classic:1.2.10",
"ch.qos.logback:logback-core:1.2.10",
@@ -116,11 +116,12 @@ http_archive(
)
http_archive(
- name = "bazel_toolchains",
- urls = ["https://github.com/bazelbuild/bazel-toolchains/archive/dac71231098d891e5c4b74a2078fe9343feef510.tar.gz"],
- strip_prefix = "bazel-toolchains-dac71231098d891e5c4b74a2078fe9343feef510",
- sha256 = "56d5370eb99559b4c74f334f81bc8a298f728bd16d5a4333c865c2ad10fae3bc",
+ name = "bazel_toolchains",
+ sha256 = "56d5370eb99559b4c74f334f81bc8a298f728bd16d5a4333c865c2ad10fae3bc",
+ strip_prefix = "bazel-toolchains-dac71231098d891e5c4b74a2078fe9343feef510",
+ urls = ["https://github.com/bazelbuild/bazel-toolchains/archive/dac71231098d891e5c4b74a2078fe9343feef510.tar.gz"],
)
load("@bazel_toolchains//repositories:repositories.bzl", bazel_toolchains_repositories = "repositories")
+
bazel_toolchains_repositories()
diff --git a/broker/BUILD.bazel b/broker/BUILD.bazel
index 51afc4f34..1c7403a44 100644
--- a/broker/BUILD.bazel
+++ b/broker/BUILD.bazel
@@ -29,7 +29,7 @@ java_library(
"//filter",
"//srvutil",
"//acl",
- "//store:io_openmessaging_storage_dledger",
+ "@maven//:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
diff --git a/container/BUILD.bazel b/container/BUILD.bazel
index 8986f7ae4..15fc0ae77 100644
--- a/container/BUILD.bazel
+++ b/container/BUILD.bazel
@@ -28,7 +28,7 @@ java_library(
"//client",
"//srvutil",
"//store",
- "//store:io_openmessaging_storage_dledger",
+ "@maven//:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:commons_collections_commons_collections",
@@ -58,7 +58,7 @@ java_library(
"//client",
"//srvutil",
"//store",
- "//store:io_openmessaging_storage_dledger",
+ "@maven//:io_openmessaging_storage_dledger",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
diff --git a/controller/BUILD.bazel b/controller/BUILD.bazel
index 0e1cc9169..ef9b9c5eb 100644
--- a/controller/BUILD.bazel
+++ b/controller/BUILD.bazel
@@ -26,7 +26,7 @@ java_library(
"//remoting",
"//client",
"//srvutil",
- "//store:io_openmessaging_storage_dledger",
+ "@maven//:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:commons_collections_commons_collections",
@@ -54,7 +54,7 @@ java_library(
"//remoting",
"//client",
"//srvutil",
- "//store:io_openmessaging_storage_dledger",
+ "@maven//:io_openmessaging_storage_dledger",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
diff --git a/store/BUILD.bazel b/store/BUILD.bazel
index fef1404de..b839e72de 100644
--- a/store/BUILD.bazel
+++ b/store/BUILD.bazel
@@ -16,57 +16,43 @@
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
-java_import(
- name = "io_openmessaging_storage_dledger",
- jars = [
- "libs/dledger-0.2.8-SNAPSHOT.jar",
- ],
- visibility = ["//visibility:public"],
-)
-
java_library(
name = "store",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
- "//logging",
"//common",
+ "//logging",
"//remoting",
- # "@maven//:io_openmessaging_storage_dledger",
- ":io_openmessaging_storage_dledger",
- "@maven//:net_java_dev_jna_jna",
- "@maven//:org_apache_commons_commons_lang3",
- "@maven//:commons_collections_commons_collections",
- "@maven//:com_conversantmedia_disruptor",
"@maven//:com_alibaba_fastjson",
- "@maven//:io_netty_netty_all",
+ "@maven//:com_conversantmedia_disruptor",
"@maven//:com_google_guava_guava",
+ "@maven//:commons_collections_commons_collections",
+ "@maven//:io_netty_netty_all",
+ "@maven//:io_openmessaging_storage_dledger",
+ "@maven//:net_java_dev_jna_jna",
+ "@maven//:org_apache_commons_commons_lang3",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
+ resources = glob(["src/test/resources/certs/*.pem"]) + glob(["src/test/resources/certs/*.key"]),
visibility = ["//visibility:public"],
deps = [
":store",
"//:test_deps",
- "//logging",
"//common",
- # "@maven//:io_openmessaging_storage_dledger",
- ":io_openmessaging_storage_dledger",
- "@maven//:org_apache_commons_commons_lang3",
- "@maven//:com_conversantmedia_disruptor",
+ "//logging",
+ "@maven//:com_conversantmedia_disruptor",
+ "@maven//:io_openmessaging_storage_dledger",
+ "@maven//:org_apache_commons_commons_lang3",
],
- resources = glob(["src/test/resources/certs/*.pem"]) + glob(["src/test/resources/certs/*.key"])
)
GenTestRules(
name = "GeneratedTestRules",
- test_files = glob(["src/test/java/**/*Test.java"]),
- deps = [
- ":tests",
- ],
exclude_tests = [
"src/test/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHATest",
"src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest",
@@ -77,4 +63,8 @@ GenTestRules(
"src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest",
"src/test/java/org/apache/rocketmq/store/HATest",
],
+ test_files = glob(["src/test/java/**/*Test.java"]),
+ deps = [
+ ":tests",
+ ],
)
diff --git a/store/libs/dledger-0.2.8-SNAPSHOT.jar b/store/libs/dledger-0.2.8-SNAPSHOT.jar
deleted file mode 100644
index d6919de58..000000000
Binary files a/store/libs/dledger-0.2.8-SNAPSHOT.jar and /dev/null differ