[Infra Improvement] Build And Run Unit Tests in Parallel (#4878)

* Support build with Bazel (#4865)

* Support build with Bazel, fixing tests to make them capable of running in concurrency and hermetic.

* Make test cases capable of running in parallel. (#4874)
This commit is contained in:
Zhanhui Li
2022-08-24 20:53:30 +08:00
committed by GitHub
parent 10d291846e
commit 3da7ea9c1b
110 changed files with 1973 additions and 412 deletions
+73
View File
@@ -0,0 +1,73 @@
#
# 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.
#
startup --host_jvm_args=-Xmx2g
run --color=yes
build --color=yes
build --enable_platform_specific_config
test --action_env=TEST_TMPDIR=/tmp
test --experimental_strict_java_deps=warn
build --experimental_strict_java_deps=warn
# This .bazelrc file contains all of the flags required for the provided
# toolchain with Remote Build Execution.
# Note your WORKSPACE must contain an rbe_autoconfig target with
# name="rbe_default" to use these flags as-is.
# Depending on how many machines are in the remote execution instance, setting
# this higher can make builds faster by allowing more jobs to run in parallel.
# Setting it too high can result in jobs that timeout, however, while waiting
# for a remote machine to execute them.
build:remote --jobs=150
# Set several flags related to specifying the platform, toolchain and java
# properties.
# These flags should only be used as is for the rbe-ubuntu16-04 container
# and need to be adapted to work with other toolchain containers.
build:remote --java_runtime_version=rbe_jdk
build:remote --tool_java_runtime_version=rbe_jdk
build:remote --extra_toolchains=@rbe_default//java:all
build:remote --crosstool_top=@rbe_default//cc:toolchain
build:remote --action_env=BAZEL_DO_NOT_DETECT_CPP_TOOLCHAIN=1
# Platform flags:
# The toolchain container used for execution is defined in the target indicated
# by "extra_execution_platforms", "host_platform" and "platforms".
# More about platforms: https://docs.bazel.build/versions/master/platforms.html
build:remote --extra_toolchains=@rbe_default//config:cc-toolchain
build:remote --extra_execution_platforms=@rbe_default//config:platform
build:remote --host_platform=@rbe_default//config:platform
build:remote --platforms=@rbe_default//config:platform
# Starting with Bazel 0.27.0 strategies do not need to be explicitly
# defined. See https://github.com/bazelbuild/bazel/issues/7480
build:remote --define=EXECUTOR=remote
# Enable remote execution so actions are performed on the remote systems.
build:remote --remote_executor=grpcs://remote.buildbuddy.io
# Enforce stricter environment rules, which eliminates some non-hermetic
# behavior and therefore improves both the remote cache hit rate and the
# correctness and repeatability of the build.
build:remote --incompatible_strict_action_env=true
# Set a higher timeout value, just in case.
build:remote --remote_timeout=3600
+1
View File
@@ -0,0 +1 @@
5.2.0
+22
View File
@@ -0,0 +1,22 @@
name: Build and Run Tests By Bazel
on:
pull_request_target:
types: [opened, reopened, synchronize]
push:
branches:
- master
- develop
- bazel
jobs:
build:
name: "Java (${{ matrix.os }})"
runs-on: ${{ matrix.os }}
strategy:
matrix:
os: [ubuntu-18.04, ubuntu-20.04, ubuntu-22.04]
steps:
- uses: actions/checkout@v2
- name: Build
run: bazel build --config=remote --remote_header=x-buildbuddy-api-key=${{ secrets.BUILD_BUDDY_API_KEY }} //...
- name: Run Tests
run: bazel test --config=remote --remote_header=x-buildbuddy-api-key=${{ secrets.BUILD_BUDDY_API_KEY }} //...
@@ -1,9 +1,9 @@
name: Build
name: Build and Run Tests By Maven
on:
pull_request:
types: [opened, reopened, synchronize]
push:
branches: [master, develop]
branches: [master, develop, bazel]
jobs:
java_build:
name: "maven-compile (${{ matrix.os }}, JDK-${{ matrix.jdk }})"
+5
View File
@@ -13,3 +13,8 @@ devenv
.DS_Store
localbin
nohup.out
bazel-out
bazel-bin
bazel-rocketmq
bazel-testlogs
.vscode
+1
View File
@@ -42,6 +42,7 @@ header:
- 'distribution/LICENSE-BIN'
- 'distribution/NOTICE-BIN'
- 'distribution/conf/rmq-proxy.json'
- '.bazelversion'
comment: on-failure
+46
View File
@@ -0,0 +1,46 @@
#
# 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.
#
load("@bazel_toolchains//rules/exec_properties:exec_properties.bzl", "create_rbe_exec_properties_dict")
platform(
name = "custom_platform",
# Inherit from the platform target generated by 'rbe_configs_gen' assuming the generated configs
# were imported as a Bazel external repository named 'rbe_default'. If you extracted the
# generated configs elsewhere in your source repository, replace the following with the label
# to the 'platform' target in the generated configs.
parents = ["@rbe_default//config:platform"],
# Example custom execution property instructing RBE to use e2-standard-2 GCE VMs.
exec_properties = create_rbe_exec_properties_dict(
container_image = "ubuntu:latest",
),
)
java_library(
name = "test_deps",
visibility = ["//visibility:public"],
exports = [
"@maven//:junit_junit",
"@maven//:org_assertj_assertj_core",
"@maven//:org_hamcrest_hamcrest_library",
"@maven//:org_mockito_mockito_core",
"@maven//:org_hamcrest_hamcrest_core",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:org_awaitility_awaitility",
"@maven//:org_openjdk_jmh_jmh_core",
"@maven//:org_openjdk_jmh_jmh_generator_annprocess",
],
)
+128
View File
@@ -0,0 +1,128 @@
#
# 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.
#
load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive")
RULES_JVM_EXTERNAL_TAG = "4.2"
RULES_JVM_EXTERNAL_SHA = "cd1a77b7b02e8e008439ca76fd34f5b07aecb8c752961f9640dea15e9e5ba1ca"
http_archive(
name = "rules_jvm_external",
sha256 = RULES_JVM_EXTERNAL_SHA,
strip_prefix = "rules_jvm_external-%s" % RULES_JVM_EXTERNAL_TAG,
url = "https://github.com/bazelbuild/rules_jvm_external/archive/%s.zip" % RULES_JVM_EXTERNAL_TAG,
)
load("@rules_jvm_external//:repositories.bzl", "rules_jvm_external_deps")
rules_jvm_external_deps()
load("@rules_jvm_external//:setup.bzl", "rules_jvm_external_setup")
rules_jvm_external_setup()
load("@rules_jvm_external//:defs.bzl", "maven_install")
maven_install(
artifacts = [
"junit:junit:4.13.2",
"com.alibaba:fastjson:1.2.76",
"org.hamcrest:hamcrest-library:1.3",
"io.netty:netty-all:4.1.65.Final",
"org.slf4j:slf4j-api:1.7.7",
"org.assertj:assertj-core:3.22.0",
"org.mockito:mockito-core:3.10.0",
"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.hamcrest:hamcrest-core:1.3",
# "io.openmessaging.storage:dledger:0.2.4",
"net.java.dev.jna:jna:4.2.2",
"ch.qos.logback:logback-classic:1.2.10",
"ch.qos.logback:logback-core:1.2.10",
"io.opentracing:opentracing-api:0.33.0",
"io.opentracing:opentracing-mock:0.33.0",
"commons-collections:commons-collections:3.2.2",
"org.awaitility:awaitility:4.1.0",
"commons-cli:commons-cli:1.4",
"com.google.guava:guava:31.0.1-jre",
"org.yaml:snakeyaml:1.30",
"commons-codec:commons-codec:1.13",
"commons-io:commons-io:2.7",
"log4j:log4j:1.2.17",
"com.google.truth:truth:0.30",
"org.bouncycastle:bcpkix-jdk15on:1.69",
"com.google.code.gson:gson:2.8.9",
"com.googlecode.concurrentlinkedhashmap:concurrentlinkedhashmap-lru:1.4.2",
"org.apache.rocketmq:rocketmq-proto:2.0.0",
"com.google.protobuf:protobuf-java:3.20.1",
"com.google.protobuf:protobuf-java-util:3.20.1",
"com.conversantmedia:disruptor:1.2.10",
"javax.annotation:javax.annotation-api:1.3.2",
"com.google.code.findbugs:jsr305:3.0.2",
"org.checkerframework:checker-qual:3.12.0",
"org.reflections:reflections:0.9.11",
"org.openjdk.jmh:jmh-core:1.19",
"org.openjdk.jmh:jmh-generator-annprocess:1.19",
"com.github.ben-manes.caffeine:caffeine:2.9.3",
"io.grpc:grpc-services:1.47.0",
"io.grpc:grpc-netty-shaded:1.47.0",
"io.grpc:grpc-context:1.47.0",
"io.grpc:grpc-stub:1.47.0",
"io.grpc:grpc-api:1.47.0",
"io.grpc:grpc-testing:1.47.0",
],
fetch_sources = True,
repositories = [
# Private repositories are supported through HTTP Basic auth
"https://repo1.maven.org/maven2",
],
)
http_archive(
name = "io_buildbuddy_buildbuddy_toolchain",
sha256 = "a2a5cccec251211e2221b1587af2ce43c36d32a42f5d881737db3b546a536510",
strip_prefix = "buildbuddy-toolchain-829c8a574f706de5c96c54ca310f139f4acda7dd",
urls = ["https://github.com/buildbuddy-io/buildbuddy-toolchain/archive/829c8a574f706de5c96c54ca310f139f4acda7dd.tar.gz"],
)
load("@io_buildbuddy_buildbuddy_toolchain//:deps.bzl", "buildbuddy_deps")
buildbuddy_deps()
load("@io_buildbuddy_buildbuddy_toolchain//:rules.bzl", "buildbuddy")
buildbuddy(name = "buildbuddy_toolchain")
load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive")
http_archive(
name = "rbe_default",
# sha256 = "c0d428774cbe70d477e1d07581d863f8dbff4ba6a66d20502d7118354a814bea",
urls = ["https://storage.googleapis.com/rbe-toolchain/bazel-configs/rbe-ubuntu1604/latest/rbe_default.tar"],
)
http_archive(
name = "bazel_toolchains",
urls = ["https://github.com/bazelbuild/bazel-toolchains/archive/dac71231098d891e5c4b74a2078fe9343feef510.tar.gz"],
strip_prefix = "bazel-toolchains-dac71231098d891e5c4b74a2078fe9343feef510",
sha256 = "56d5370eb99559b4c74f334f81bc8a298f728bd16d5a4333c865c2ad10fae3bc",
)
load("@bazel_toolchains//repositories:repositories.bzl", bazel_toolchains_repositories = "repositories")
bazel_toolchains_repositories()
+72
View File
@@ -0,0 +1,72 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "acl",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//common",
"//remoting",
"//logging",
"//srvutil",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:org_yaml_snakeyaml",
"@maven//:commons_codec_commons_codec",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:com_google_protobuf_protobuf_java",
"@maven//:com_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":acl",
"//common",
"//remoting",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:org_yaml_snakeyaml",
"@maven//:commons_codec_commons_codec",
"@maven//:com_alibaba_fastjson",
],
resources = glob(["src/test/resources/**/*.yml"])
)
GenTestRules(
name = "GeneratedTestRules",
test_files = glob(["src/test/java/**/*Test.java"]),
deps = [
":tests",
],
# The following tests are not hermetic. Fix them later.
exclude_tests = [
"src/test/java/org/apache/rocketmq/acl/plain/PlainAccessControlFlowTest",
"src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionManagerTest",
"src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest",
],
)
@@ -17,14 +17,11 @@
package org.apache.rocketmq.acl.common;
import com.alibaba.fastjson.JSONObject;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.nio.channels.FileChannel;
import java.util.Map;
import java.util.SortedMap;
import org.apache.commons.lang3.StringUtils;
@@ -239,82 +236,38 @@ public class AclUtils {
}
public static <T> T getYamlDataObject(String path, Class<T> clazz) {
Yaml yaml = new Yaml();
FileInputStream fis = null;
try {
fis = new FileInputStream(new File(path));
return yaml.loadAs(fis, clazz);
try (FileInputStream fis = new FileInputStream(path)) {
return getYamlDataObject(fis, clazz);
} catch (FileNotFoundException ignore) {
return null;
} catch (Exception e) {
throw new AclException(e.getMessage());
} finally {
if (fis != null) {
try {
fis.close();
} catch (IOException ignore) {
}
}
throw new AclException(e.getMessage(), e);
}
}
public static <T> T getYamlDataObject(InputStream fis, Class<T> clazz) {
Yaml yaml = new Yaml();
try {
return yaml.loadAs(fis, clazz);
} catch (Exception e) {
throw new AclException(e.getMessage(), e);
}
}
public static boolean writeDataObject(String path, Map<String, Object> dataMap) {
Yaml yaml = new Yaml();
PrintWriter pw = null;
try {
pw = new PrintWriter(new FileWriter(path));
try (PrintWriter pw = new PrintWriter(new FileWriter(path))) {
String dumpAsMap = yaml.dumpAsMap(dataMap);
pw.print(dumpAsMap);
pw.flush();
} catch (Exception e) {
throw new AclException(e.getMessage());
} finally {
if (pw != null) {
pw.close();
}
throw new AclException(e.getMessage(), e);
}
return true;
}
public static boolean copyFile(String from, String to) {
FileChannel input = null;
FileChannel output = null;
try {
input = new FileInputStream(new File(from)).getChannel();
output = new FileOutputStream(new File(to)).getChannel();
output.transferFrom(input, 0, input.size());
return true;
} catch (Exception e) {
log.error("file copy error. from={}, to={}", from, to, e);
} finally {
closeFileChannel(input);
closeFileChannel(output);
}
return false;
}
public static boolean moveFile(String from, String to) {
try {
File file = new File(from);
return file.renameTo(new File(to));
} catch (Exception e) {
log.error("file move error. from={}, to={}", from, to, e);
}
return false;
}
private static void closeFileChannel(FileChannel fileChannel) {
if (fileChannel != null) {
try {
fileChannel.close();
} catch (IOException e) {
log.error("Close file channel error.", e);
}
}
}
public static RPCHook getAclRPCHook(String fileName) {
JSONObject yamlDataObject = null;
JSONObject yamlDataObject;
try {
yamlDataObject = AclUtils.getYamlDataObject(fileName,
JSONObject.class);
@@ -322,9 +275,23 @@ public class AclUtils {
log.error("Convert yaml file to data object error, ", e);
return null;
}
return buildRpcHook(yamlDataObject);
}
public static RPCHook getAclRPCHook(InputStream inputStream) {
JSONObject yamlDataObject = null;
try {
yamlDataObject = AclUtils.getYamlDataObject(inputStream, JSONObject.class);
} catch (Exception e) {
log.error("Convert yaml file to data object error, ", e);
return null;
}
return buildRpcHook(yamlDataObject);
}
private static RPCHook buildRpcHook(JSONObject yamlDataObject) {
if (yamlDataObject == null || yamlDataObject.isEmpty()) {
log.warn("Cannot find conf file :{}, acl isn't be enabled.", fileName);
log.warn("Failed to parse configuration to enable ACL.");
return null;
}
@@ -332,8 +299,7 @@ public class AclUtils {
String secretKey = yamlDataObject.getString(AclConstants.CONFIG_SECRET_KEY);
if (StringUtils.isBlank(accessKey) || StringUtils.isBlank(secretKey)) {
log.warn("AccessKey or secretKey is blank, the acl is not enabled.");
log.warn("Failed to enable ACL. Either AccessKey or secretKey is blank");
return null;
}
return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
@@ -50,18 +50,18 @@ public class RemoteAddressStrategyFactory {
String[] strArray = StringUtils.split(remoteAddr, ":");
String last = strArray[strArray.length - 1];
if (!last.startsWith("{")) {
throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress: %s", remoteAddr));
}
return new MultipleRemoteAddressStrategy(AclUtils.getAddresses(remoteAddr, last));
} else {
String[] strArray = StringUtils.split(remoteAddr, ".");
// However a right IP String provided by user,it always can be divided into 4 parts by '.'.
if (strArray.length < 4) {
throw new AclException(String.format("MultipleRemoteAddressStrategy has got a/some wrong format IP(s) ", remoteAddr));
throw new AclException(String.format("MultipleRemoteAddressStrategy has got a/some wrong format IP(s): %s ", remoteAddr));
}
String lastStr = strArray[strArray.length - 1];
if (!lastStr.startsWith("{")) {
throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress: %s", remoteAddr));
}
return new MultipleRemoteAddressStrategy(AclUtils.getAddresses(remoteAddr, lastStr));
}
@@ -19,11 +19,15 @@ package org.apache.rocketmq.acl.common;
import com.alibaba.fastjson.JSONObject;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.remoting.RPCHook;
import org.junit.Assert;
@@ -32,13 +36,10 @@ import org.junit.Test;
public class AclUtilsTest {
@Test
public void getAddresses() {
public void testGetAddresses() {
String address = "1.1.1.{1,2,3,4}";
String[] addressArray = AclUtils.getAddresses(address, "{1,2,3,4}");
List<String> newAddressList = new ArrayList<>();
for (String a : addressArray) {
newAddressList.add(a);
}
List<String> newAddressList = new ArrayList<>(Arrays.asList(addressArray));
List<String> addressList = new ArrayList<>();
addressList.add("1.1.1.1");
@@ -47,13 +48,11 @@ public class AclUtilsTest {
addressList.add("1.1.1.4");
Assert.assertEquals(newAddressList, addressList);
// IPv6 test
// IPv6 test
String ipv6Address = "1:ac41:9987::bb22:666:{1,2,3,4}";
String[] ipv6AddressArray = AclUtils.getAddresses(ipv6Address, "{1,2,3,4}");
List<String> newIPv6AddressList = new ArrayList<>();
for (String a : ipv6AddressArray) {
newIPv6AddressList.add(a);
}
Collections.addAll(newIPv6AddressList, ipv6AddressArray);
List<String> ipv6AddressList = new ArrayList<>();
ipv6AddressList.add("1:ac41:9987::bb22:666:1");
@@ -64,7 +63,7 @@ public class AclUtilsTest {
}
@Test
public void isScopeStringArray() {
public void testIsScope_StringArray() {
String address = "12";
for (int i = 0; i < 6; i++) {
@@ -79,7 +78,7 @@ public class AclUtilsTest {
}
@Test
public void isScopeArray() {
public void testIsScope_Array() {
String[] address = StringUtils.split("12.12.12.12", ".");
boolean isScope = AclUtils.isScope(address, 4);
Assert.assertTrue(isScope);
@@ -92,7 +91,7 @@ public class AclUtilsTest {
isScope = AclUtils.isScope(address, 3);
Assert.assertFalse(isScope);
// IPv6 test
// IPv6 test
address = StringUtils.split("1050:0000:0000:0000:0005:0600:300c:326b", ":");
isScope = AclUtils.isIPv6Scope(address, 8);
Assert.assertTrue(isScope);
@@ -110,11 +109,10 @@ public class AclUtilsTest {
Assert.assertFalse(isScope);
isScope = AclUtils.isIPv6Scope(address, 4);
Assert.assertTrue(isScope);
}
@Test
public void isScopeStringTest() {
public void testIsScope_String() {
for (int i = 0; i < 256; i++) {
boolean isScope = AclUtils.isScope(i + "");
Assert.assertTrue(isScope);
@@ -126,7 +124,7 @@ public class AclUtilsTest {
}
@Test
public void isScopeTest() {
public void testIsScope_Integral() {
for (int i = 0; i < 256; i++) {
boolean isScope = AclUtils.isScope(i);
Assert.assertTrue(isScope);
@@ -136,7 +134,7 @@ public class AclUtilsTest {
isScope = AclUtils.isScope(256);
Assert.assertFalse(isScope);
// IPv6 test
// IPv6 test
int min = Integer.parseInt("0", 16);
int max = Integer.parseInt("ffff", 16);
for (int i = min; i < max + 1; i++) {
@@ -147,11 +145,10 @@ public class AclUtilsTest {
Assert.assertFalse(isScope);
isScope = AclUtils.isIPv6Scope(max + 1);
Assert.assertFalse(isScope);
}
@Test
public void isAsteriskTest() {
public void testIsAsterisk() {
boolean isAsterisk = AclUtils.isAsterisk("*");
Assert.assertTrue(isAsterisk);
@@ -160,7 +157,7 @@ public class AclUtilsTest {
}
@Test
public void isColonTest() {
public void testIsComma() {
boolean isColon = AclUtils.isComma(",");
Assert.assertTrue(isColon);
@@ -169,7 +166,7 @@ public class AclUtilsTest {
}
@Test
public void isMinusTest() {
public void testIsMinus() {
boolean isMinus = AclUtils.isMinus("-");
Assert.assertTrue(isMinus);
@@ -178,30 +175,21 @@ public class AclUtilsTest {
}
@Test
public void v6ipProcessTest() {
public void testV6ipProcess() {
String remoteAddr = "5::7:6:1-200:*";
String[] strArray = StringUtils.split(remoteAddr, ":");
Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr), "0005:0000:0000:0000:0007:0006");
// Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0000:0000:0000:0007:0006");
remoteAddr = "5::7:6:1-200";
strArray = StringUtils.split(remoteAddr, ":");
Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr), "0005:0000:0000:0000:0000:0007:0006");
// Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0000:0000:0000:0000:0007:0006");
remoteAddr = "5::7:6:*";
strArray = StringUtils.split(remoteAddr, ":");
Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr), "0005:0000:0000:0000:0000:0007:0006");
// Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0000:0000:0000:0000:0007:0006");
remoteAddr = "5:7:6:*";
strArray = StringUtils.split(remoteAddr, ":");
Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr), "0005:0007:0006");
// Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0007:0006");
}
@Test
public void expandIPTest() {
public void testExpandIP() {
Assert.assertEquals(AclUtils.expandIP("::", 8), "0000:0000:0000:0000:0000:0000:0000:0000");
Assert.assertEquals(AclUtils.expandIP("::1", 8), "0000:0000:0000:0000:0000:0000:0000:0001");
Assert.assertEquals(AclUtils.expandIP("3::", 8), "0003:0000:0000:0000:0000:0000:0000:0000");
@@ -214,19 +202,30 @@ public class AclUtilsTest {
@SuppressWarnings("unchecked")
@Test
public void getYamlDataObjectTest() {
public void testGetYamlDataObject() throws IOException {
try (InputStream is = AclUtilsTest.class.getClassLoader().getResourceAsStream("conf/plain_acl_correct.yml")) {
Map<String, Object> map = AclUtils.getYamlDataObject(is, Map.class);
Assert.assertNotNull(map);
Assert.assertFalse(map.isEmpty());
}
}
Map<String, Object> map = AclUtils.getYamlDataObject("src/test/resources/conf/plain_acl_correct.yml", Map.class);
Assert.assertFalse(map.isEmpty());
private static String randomTmpFile() {
String tmpFileName = System.getProperty("java.io.tmpdir");
// https://rationalpi.wordpress.com/2007/01/26/javaiotmpdir-inconsitency/
if (!tmpFileName.endsWith(File.separator)) {
tmpFileName += File.separator;
}
tmpFileName += UUID.randomUUID() + ".yml";
return tmpFileName;
}
@Test
public void writeDataObject2YamlFileTest() throws IOException {
String targetFileName = "src/test/resources/conf/plain_write_acl.yml";
String targetFileName = randomTmpFile();
File transport = new File(targetFileName);
transport.delete();
transport.createNewFile();
Assert.assertTrue(transport.createNewFile());
transport.deleteOnExit();
Map<String, Object> aclYamlMap = new HashMap<String, Object>();
@@ -249,17 +248,14 @@ public class AclUtilsTest {
accounts.add(accountsMap);
aclYamlMap.put("accounts", accounts);
Assert.assertTrue(AclUtils.writeDataObject(targetFileName, aclYamlMap));
transport.delete();
}
@Test
public void updateExistedYamlFileTest() throws IOException {
String targetFileName = "src/test/resources/conf/plain_update_acl.yml";
String targetFileName = randomTmpFile();
File transport = new File(targetFileName);
transport.delete();
transport.createNewFile();
Assert.assertTrue(transport.createNewFile());
transport.deleteOnExit();
Map<String, Object> aclYamlMap = new HashMap<String, Object>();
@@ -283,53 +279,20 @@ public class AclUtilsTest {
Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
List<String> updatedGlobalWhiteRemoteAddrs = (List<String>) readableMap.get("globalWhiteRemoteAddrs");
Assert.assertEquals("192.168.1.2", updatedGlobalWhiteRemoteAddrs.get(0));
transport.delete();
}
@Test
public void getYamlDataIgnoreFileNotFoundExceptionTest() {
JSONObject yamlDataObject = AclUtils.getYamlDataObject("plain_acl.yml", JSONObject.class);
Assert.assertTrue(yamlDataObject == null);
}
@Test
public void getAclRPCHookTest() {
//RPCHook errorContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_format_error.yml");
//Assert.assertNull(errorContRPCHook);
RPCHook noFileRPCHook = AclUtils.getAclRPCHook("src/test/resources/plain_acl_format_error1.yml");
Assert.assertNull(noFileRPCHook);
//RPCHook emptyContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_null.yml");
//Assert.assertNull(emptyContRPCHook);
RPCHook incompleteContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_incomplete.yml");
Assert.assertNull(incompleteContRPCHook);
Assert.assertNull(yamlDataObject);
}
@Test
public void testCopyAndMoveFile() {
String path = "src/test/resources/conf/plain_acl.yml";
String backupPath = "src/test/resources/conf/plain_acl.yml_backup";
Map<String, Object> aclYamlData = AclUtils.getYamlDataObject(path, Map.class);
AclUtils.copyFile(path, backupPath);
Assert.assertEquals(aclYamlData, AclUtils.getYamlDataObject(backupPath, Map.class));
Map<String, Object> aclYamlMap = new HashMap<String, Object>();
List<String> globalWhiteRemoteAddrs = new ArrayList<String>();
globalWhiteRemoteAddrs.add("10.10.103.*");
globalWhiteRemoteAddrs.add("192.168.0.*");
aclYamlMap.put("globalWhiteRemoteAddrs", globalWhiteRemoteAddrs);
AclUtils.writeDataObject(path, aclYamlMap);
Assert.assertNotEquals(aclYamlData, AclUtils.getYamlDataObject(path, Map.class));
AclUtils.moveFile(backupPath, path);
Assert.assertEquals(aclYamlData, AclUtils.getYamlDataObject(path, Map.class));
public void getAclRPCHookTest() throws IOException {
try (InputStream is = AclUtilsTest.class.getClassLoader().getResourceAsStream("conf/plain_acl_incomplete.yml")) {
RPCHook incompleteContRPCHook = AclUtils.getAclRPCHook(is);
Assert.assertNull(incompleteContRPCHook);
}
}
}
+16
View File
@@ -0,0 +1,16 @@
#
# 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.
#
+109
View File
@@ -0,0 +1,109 @@
#
# 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.
#
"""Generate java test rules from given test_files.
Instead of having to create one test rule per test in the BUILD file, this rule
provides a handy way to create a bunch of test rules for the specified test
files.
"""
def GenTestRules(
name,
test_files,
deps,
exclude_tests = [],
default_test_size = "small",
small_tests = [],
medium_tests = [],
large_tests = [],
enormous_tests = [],
resources = [],
flaky_tests = [],
tags = [],
prefix = "",
jvm_flags = [],
args = [],
visibility = None,
shard_count = 1):
for test in _get_test_names(test_files):
if test in exclude_tests:
continue
test_size = default_test_size
if test in small_tests:
test_size = "small"
if test in medium_tests:
test_size = "medium"
if test in large_tests:
test_size = "large"
if test in enormous_tests:
test_size = "enormous"
flaky = 0
if (test in flaky_tests) or ("flaky" in tags):
flaky = 1
java_class = _package_from_path(
native.package_name() + "/" + _strip_right(test, ".java"),
)
package = java_class[:java_class.rfind(".")]
native.java_test(
name = prefix + test,
runtime_deps = deps,
resources = resources,
size = test_size,
jvm_flags = jvm_flags,
args = args,
flaky = flaky,
tags = tags,
test_class = java_class,
visibility = visibility,
shard_count = shard_count,
)
def _get_test_names(test_files):
test_names = []
for test_file in test_files:
if not test_file.endswith("Test.java") and not test_file.endswith("IT.java"):
continue
test_names += [test_file[:-5]]
return test_names
def _package_from_path(package_path, src_impls = None):
src_impls = src_impls or ["javatests/", "java/"]
for src_impl in src_impls:
if not src_impl.endswith("/"):
src_impl += "/"
index = _index_of_end(package_path, src_impl)
if index >= 0:
package_path = package_path[index:]
break
return package_path.replace("/", ".")
def _strip_right(str, suffix):
"""Returns str without the suffix if it ends with suffix."""
if str.endswith(suffix):
return str[0:len(str) - len(suffix)]
else:
return str
def _index_of_end(str, part):
"""If part is in str, return the index of the first character after part.
Return -1 if part is not in str."""
index = str.find(part)
if index >= 0:
return index + len(part)
return -1
+82
View File
@@ -0,0 +1,82 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "broker",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//remoting",
"//logging",
"//common",
"//store",
"//client",
"//filter",
"//srvutil",
"//acl",
"//store:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:org_slf4j_slf4j_api",
"@maven//:commons_cli_commons_cli",
"@maven//:com_google_guava_guava",
"@maven//:com_googlecode_concurrentlinkedhashmap_concurrentlinkedhashmap_lru",
"@maven//:commons_io_commons_io",
"@maven//:commons_collections_commons_collections",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":broker",
"//acl",
"//client",
"//filter",
"//logging",
"//store",
"//common",
"//remoting",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
],
resources = [
"src/test/resources/logback-test.xml",
"src/test/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator",
"src/test/resources/META-INF/service/org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener",
"src/test/resources/META-INF/service/org.apache.rocketmq.broker.transaction.TransactionalMessageService",
],
)
GenTestRules(
name = "GeneratedTestRules",
test_files = glob(["src/test/java/**/*Test.java"]),
deps = [
":tests",
],
)
@@ -415,7 +415,13 @@ public class BrokerController {
protected void initializeRemotingServer() throws CloneNotSupportedException {
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
int listeningPort = nettyServerConfig.getListenPort() - 2;
if (listeningPort < 0) {
listeningPort = 0;
}
fastConfig.setListenPort(listeningPort);
this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
}
@@ -1427,12 +1433,19 @@ public class BrokerController {
if (this.remotingServer != null) {
this.remotingServer.start();
// In test scenarios where it is up to OS to pick up an available port, set the listening port back to config
if (null != nettyServerConfig && 0 == nettyServerConfig.getListenPort()) {
nettyServerConfig.setListenPort(remotingServer.localListenPort());
}
}
if (this.fastRemotingServer != null) {
this.fastRemotingServer.start();
}
this.storeHost = new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort());
for (BrokerAttachedPlugin brokerAttachedPlugin : brokerAttachedPlugins) {
if (brokerAttachedPlugin != null) {
brokerAttachedPlugin.start();
@@ -70,15 +70,12 @@ public abstract class AbstractSendMessageProcessor implements NettyRequestProces
protected List<ConsumeMessageHook> consumeMessageHookList;
protected final static int DLQ_NUMS_PER_GROUP = 1;
protected final Random random = new Random(System.currentTimeMillis());
protected final SocketAddress storeHost;
private List<SendMessageHook> sendMessageHookList;
protected final BrokerController brokerController;
protected final Random random = new Random(System.currentTimeMillis());
private List<SendMessageHook> sendMessageHookList;
public AbstractSendMessageProcessor(final BrokerController brokerController) {
this.brokerController = brokerController;
this.storeHost = brokerController.getStoreHost();
}
public void registerConsumeMessageHook(List<ConsumeMessageHook> consumeMessageHookList) {
@@ -126,7 +123,7 @@ public abstract class AbstractSendMessageProcessor implements NettyRequestProces
}
String newTopic = MixAll.getRetryTopic(requestHeader.getGroup());
int queueIdInt = Math.abs(this.random.nextInt() % 99999999) % subscriptionGroupConfig.getRetryQueueNums();
int queueIdInt = this.random.nextInt(subscriptionGroupConfig.getRetryQueueNums());
int topicSysFlag = 0;
if (requestHeader.isUnitMode()) {
@@ -293,7 +290,7 @@ public abstract class AbstractSendMessageProcessor implements NettyRequestProces
requestHeader.getOriginTopic(),
requestHeader.getGroup(),
uniqKey,
putMessageResult == null ? "null" : putMessageResult.getPutMessageStatus().toString());
"null");
}
response.setCode(ResponseCode.SYSTEM_ERROR);
@@ -419,7 +416,7 @@ public abstract class AbstractSendMessageProcessor implements NettyRequestProces
}
public SocketAddress getStoreHost() {
return storeHost;
return brokerController.getStoreHost();
}
protected RemotingCommand msgContentCheck(final ChannelHandlerContext ctx,
@@ -18,6 +18,7 @@
package org.apache.rocketmq.broker;
import java.io.File;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -30,19 +31,37 @@ import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.netty.RequestTask;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
public class BrokerControllerTest {
private MessageStoreConfig messageStoreConfig;
private BrokerConfig brokerConfig;
private NettyServerConfig nettyServerConfig;
@Before
public void setUp() {
messageStoreConfig = new MessageStoreConfig();
String storePathRootDir = System.getProperty("java.io.tmpdir") + File.separator + "store-"
+ UUID.randomUUID().toString();
messageStoreConfig.setStorePathRootDir(storePathRootDir);
brokerConfig = new BrokerConfig();
nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(0);
}
@Test
public void testBrokerRestart() throws Exception {
BrokerController brokerController = new BrokerController(
new BrokerConfig(),
new NettyServerConfig(),
new NettyClientConfig(),
new MessageStoreConfig());
BrokerController brokerController = new BrokerController(brokerConfig, nettyServerConfig, new NettyClientConfig(), messageStoreConfig);
assertThat(brokerController.initialize()).isTrue();
brokerController.start();
brokerController.shutdown();
@@ -50,16 +69,12 @@ public class BrokerControllerTest {
@After
public void destroy() {
UtilAll.deleteFile(new File(new MessageStoreConfig().getStorePathRootDir()));
UtilAll.deleteFile(new File(messageStoreConfig.getStorePathRootDir()));
}
@Test
public void testHeadSlowTimeMills() throws Exception {
BrokerController brokerController = new BrokerController(
new BrokerConfig(),
new NettyServerConfig(),
new NettyClientConfig(),
new MessageStoreConfig());
BrokerController brokerController = new BrokerController(brokerConfig, nettyServerConfig, new NettyClientConfig(), messageStoreConfig);
brokerController.initialize();
BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
@@ -61,7 +61,7 @@ public class MessageStoreWithFilterTest {
private static final String topic = "topic";
private static final int queueId = 0;
private static final String storePath = "." + File.separator + "unit_test_store";
private static final String storePath = System.getProperty("java.io.tmpdir") + File.separator + "unit_test_store";
private static final int commitLogFileSize = 1024 * 1024 * 256;
private static final int cqFileSize = 300000 * 20;
private static final int cqExtFileSize = 300000 * 128;
@@ -44,12 +44,12 @@ public class ConsumerOffsetManagerTest {
@Test
public void cleanOffsetByTopic_NotExist(){
consumerOffsetManager.cleanOffsetByTopic("InvalidTopic");
assertThat(consumerOffsetManager.getOffsetTable().containsKey(key));
assertThat(consumerOffsetManager.getOffsetTable().containsKey(key)).isTrue();
}
@Test
public void cleanOffsetByTopic_Exist(){
consumerOffsetManager.cleanOffsetByTopic("FooBar");
assertThat(!consumerOffsetManager.getOffsetTable().containsKey(key));
assertThat(!consumerOffsetManager.getOffsetTable().containsKey(key)).isTrue();
}
}
@@ -73,7 +73,7 @@ public class ScheduleMessageServiceTest {
*/
int delayLevel = 3;
private static final String storePath = System.getProperty("user.home") + File.separator + "schedule_test#" + UUID.randomUUID();
private static final String storePath = System.getProperty("java.io.tmpdir") + File.separator + "schedule_test#" + UUID.randomUUID();
private static final int commitLogFileSize = 1024;
private static final int cqFileSize = 10;
private static final int cqExtFileSize = 10 * (ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE + 64);
@@ -113,6 +113,8 @@ public class ScheduleMessageServiceTest {
messageStoreConfig.setEnableConsumeQueueExt(true);
messageStoreConfig.setStorePathRootDir(storePath);
messageStoreConfig.setStorePathCommitLog(storePath + File.separator + "commitlog");
// Let OS pick an available port
messageStoreConfig.setHaListenPort(0);
brokerConfig = new BrokerConfig();
BrokerStatsManager manager = new BrokerStatsManager(brokerConfig.getBrokerClusterName(), brokerConfig.isEnableDetailStat());
+65
View File
@@ -0,0 +1,65 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "client",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//common",
"//remoting",
"//logging",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:io_opentracing_opentracing_api",
"@maven//:commons_collections_commons_collections",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":client",
"//remoting",
"//common",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:io_opentracing_opentracing_api",
"@maven//:io_opentracing_opentracing_mock",
"@maven//:org_awaitility_awaitility",
],
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",
],
medium_tests = [
"src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest",
],
)
@@ -1051,7 +1051,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
Validators.checkMessage(msg, defaultMQProducer);
if (!msg.getTopic().equals(mq.getTopic())) {
throw new MQClientException("message's topic not equal mq's topic", null);
throw new MQClientException("Topic of the message does not match its target message queue", null);
}
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout > costTime) {
@@ -53,6 +53,7 @@ import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.RPCHook;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
@@ -93,6 +94,11 @@ public class DefaultLitePullConsumerTest {
private String brokerName = "BrokerA";
private boolean flag = false;
@BeforeClass
public static void setEnv() {
System.setProperty("rocketmq.client.logRoot", System.getProperty("java.io.tmpdir"));
}
@Before
public void init() throws Exception {
ConcurrentMap<String, MQClientInstance> factoryTable = (ConcurrentMap<String, MQClientInstance>) FieldUtils.readDeclaredField(MQClientManager.getInstance(), "factoryTable", true);
@@ -161,7 +161,7 @@ public class MQClientInstanceTest {
runningInfo = mqClientInstance.consumerRunningInfo(group);
assertThat(runningInfo).isNotNull();
assertThat(mockConsumerInner.consumerRunningInfo().getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE));
assertThat(mockConsumerInner.consumerRunningInfo().getProperties().get(ConsumerRunningInfo.PROP_CONSUME_TYPE)).isNotNull();
mqClientInstance.unregisterConsumer(group);
flag = mqClientInstance.registerConsumer(group, mock(MQConsumerInner.class));
@@ -53,6 +53,7 @@ import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.RPCHook;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
@@ -104,6 +105,11 @@ public class DefaultMQLitePullConsumerWithTraceTest {
private String customerTraceTopic = "rmq_trace_topic_12345";
@BeforeClass
public static void setUpEnv() {
System.setProperty("rocketmq.client.logRoot", System.getProperty("java.io.tmpdir"));
}
@Before
public void init() throws Exception {
Field field = MQClientInstance.class.getDeclaredField("rebalanceService");
+60
View File
@@ -0,0 +1,60 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "common",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//remoting",
"//logging",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:commons_collections_commons_collections",
"@maven//:commons_codec_commons_codec",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":common",
"//remoting",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
],
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",
],
)
@@ -220,7 +220,7 @@ public class MixAll {
}
if (result) {
return new String(data);
return new String(data, "UTF-8");
}
}
return null;
@@ -75,7 +75,7 @@ public class MixAllTest {
file.delete();
}
file.createNewFile();
PrintWriter out = new PrintWriter(fileName);
PrintWriter out = new PrintWriter(fileName, "UTF-8");
out.write("TestForMixAll_中文");
out.close();
String string = MixAll.file2String(fileName);
@@ -29,7 +29,7 @@ public class MessageClientIDSetterTest {
long t = System.currentTimeMillis();
String uniqID = MessageClientIDSetter.createUniqID();
long t2 = MessageClientIDSetter.getNearlyTimeFromID(uniqID).getTime();
assertThat(t2 - t < 20);
assertThat(t2 - t < 20).isTrue();
}
@Test
@@ -40,7 +40,7 @@ public class MessageClientIDSetterTest {
String idHex2 = uniqID2.substring(uniqID2.length() - 4);
int s1 = Integer.parseInt(idHex, 16);
int s2 = Integer.parseInt(idHex2, 16);
assertThat(s1 == s2 - 1);
assertThat(s1 == s2 - 1).isTrue();
}
@@ -257,6 +257,7 @@ public class MessageDecoderTest {
assertThat("abc").isEqualTo(decodedMsg.getTopic());
}
@Test
public void testNullValueProperty() throws Exception {
MessageExt msg = new MessageExt();
msg.setBody("x".getBytes());
@@ -32,7 +32,10 @@ import java.util.List;
public class IOTinyUtilsTest {
private String testRootDir = System.getProperty("user.home") + File.separator + "iotinyutilstest";
/**
* https://bazel.build/reference/test-encyclopedia#filesystem
*/
private String testRootDir = System.getProperty("java.io.tmpdir") + File.separator + "iotinyutilstest";
@Before
public void init() {
@@ -50,7 +53,6 @@ public class IOTinyUtilsTest {
UtilAll.deleteFile(file);
}
@Test
public void testToString() throws Exception {
byte[] b = "testToString".getBytes(RemotingHelper.DEFAULT_CHARSET);
@@ -150,7 +152,7 @@ public class IOTinyUtilsTest {
@Test
public void testCopyFile() throws Exception {
File source = new File(testRootDir, "soruce");
File source = new File(testRootDir, "source");
String target = testRootDir + File.separator + "dest";
IOTinyUtils.writeStringToFile(source, "testCopyFile", RemotingHelper.DEFAULT_CHARSET);
+82
View File
@@ -0,0 +1,82 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "container",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//broker",
"//common",
"//logging",
"//remoting",
"//client",
"//srvutil",
"//store",
"//store:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:commons_collections_commons_collections",
"@maven//:commons_codec_commons_codec",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_core",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:commons_cli_commons_cli",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":container",
"//broker",
"//common",
"//logging",
"//remoting",
"//client",
"//srvutil",
"//store",
"//store:io_openmessaging_storage_dledger",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
],
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",
],
# The following tests are flaky, fix them later.
exclude_tests = [
"src/test/java/org/apache/rocketmq/container/BrokerContainerStartupTest",
"src/test/java/org/apache/rocketmq/container/BrokerContainerTest",
],
)
+76
View File
@@ -0,0 +1,76 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "controller",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//common",
"//logging",
"//remoting",
"//client",
"//srvutil",
"//store:io_openmessaging_storage_dledger",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:commons_collections_commons_collections",
"@maven//:commons_codec_commons_codec",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_core",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:commons_cli_commons_cli",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":controller",
"//common",
"//logging",
"//remoting",
"//client",
"//srvutil",
"//store:io_openmessaging_storage_dledger",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
],
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",
],
medium_tests = [
"src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest",
],
)
@@ -54,8 +54,9 @@ public class DLedgerControllerTest {
private List<String> baseDirs;
private List<DLedgerController> controllers;
public DLedgerController launchController(final String group, final String peers, final String selfId, String storeType, final boolean isEnableElectUncleanMaster) {
final String path = "/tmp" + File.separator + group + File.separator + selfId;
public DLedgerController launchController(final String group, final String peers, final String selfId,
String storeType, final boolean isEnableElectUncleanMaster) {
final String path = System.getProperty("java.io.tmpdir") + File.separator + group + File.separator + selfId;
baseDirs.add(path);
final ControllerConfig config = new ControllerConfig();
+59
View File
@@ -0,0 +1,59 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "filter",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//common",
"//remoting",
"//logging",
"//srvutil",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":filter",
"//common",
"//remoting",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
],
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",
],
)
+24
View File
@@ -0,0 +1,24 @@
#
# 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.
#
java_library(
name = "logging",
srcs = glob(["src/main/java/**/*.java"]),
deps = [
"@maven//:org_slf4j_slf4j_api",
],
visibility = ["//visibility:public"],
)
+76
View File
@@ -0,0 +1,76 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "namesrv",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//remoting",
"//logging",
"//srvutil",
"//tools",
"//client",
"//common",
"//controller",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:ch_qos_logback_logback_core",
"@maven//:org_slf4j_slf4j_api",
"@maven//:org_bouncycastle_bcpkix_jdk15on",
"@maven//:commons_cli_commons_cli",
"@maven//:com_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":namesrv",
"//remoting",
"//logging",
"//srvutil",
"//tools",
"//client",
"//common",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_cli_commons_cli",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_core",
],
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",
],
)
@@ -250,6 +250,11 @@ public class NamesrvController {
this.remotingServer.start();
this.remotingClient.start();
// In test scenarios where it is up to OS to pick up an available port, set the listening port back to config
if (0 == nettyServerConfig.getListenPort()) {
nettyServerConfig.setListenPort(this.remotingServer.localListenPort());
}
if (this.fileWatchService != null) {
this.fileWatchService.start();
}
@@ -18,10 +18,8 @@ package org.apache.rocketmq.namesrv;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.Callable;
@@ -84,7 +82,7 @@ public class NamesrvStartup {
return null;
}
public static void parseCommandlineAndConfigFile(String[] args) throws IOException, JoranException {
public static void parseCommandlineAndConfigFile(String[] args) throws Exception {
System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
//PackageConflictDetect.detectFastjson();
@@ -20,6 +20,7 @@ import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -37,6 +38,14 @@ public class NameServerInstanceTest {
nameSrvController.start();
}
/**
* Add a placeholder to make Bazel happy.
*/
@Test
public void itWorks() {
}
@After
public void shutdown() throws Exception {
if (nameSrvController != null) {
+97
View File
@@ -0,0 +1,97 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "proxy",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//remoting",
"//logging",
"//common",
"//client",
"//broker",
"//acl",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:commons_collections_commons_collections",
"@maven//:commons_codec_commons_codec",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_github_ben_manes_caffeine_caffeine",
"@maven//:io_grpc_grpc_services",
"@maven//:com_google_protobuf_protobuf_java",
"@maven//:com_google_protobuf_protobuf_java_util",
"@maven//:io_grpc_grpc_netty_shaded",
"@maven//:io_grpc_grpc_context",
"@maven//:io_grpc_grpc_stub",
"@maven//:io_grpc_grpc_api",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_core",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:com_google_code_findbugs_jsr305",
"@maven//:org_checkerframework_checker_qual",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":proxy",
"//remoting",
"//broker",
"//client",
"//common",
"//:test_deps",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_core",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
"@maven//:com_google_protobuf_protobuf_java",
"@maven//:com_google_protobuf_protobuf_java_util",
"@maven//:io_grpc_grpc_netty_shaded",
"@maven//:io_grpc_grpc_context",
"@maven//:io_grpc_grpc_stub",
"@maven//:io_grpc_grpc_api",
],
resources = [
"src/test/resources/rmq-proxy-home/conf/broker.conf",
"src/test/resources/rmq-proxy-home/conf/rmq-proxy.json",
"src/test/resources/rmq-proxy-home/conf/logback_proxy.xml",
],
)
GenTestRules(
name = "GeneratedTestRules",
test_files = glob(["src/test/java/**/*Test.java"]),
deps = [
":tests",
],
exclude_tests = [
"src/test/java/org/apache/rocketmq/proxy/config/InitConfigAndLoggerTest",
],
)
@@ -18,7 +18,11 @@
package org.apache.rocketmq.proxy.config;
import com.alibaba.fastjson.JSON;
import com.google.common.base.Charsets;
import com.google.common.io.CharStreams;
import java.io.File;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.common.constant.LoggerName;
@@ -41,6 +45,13 @@ public class Configuration {
}
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));
}
}
String filePath = new File(ConfigurationManager.getProxyHome() + File.separator + "conf", configFileName).toString();
File file = new File(filePath);
@@ -77,8 +77,7 @@ public class GrpcClientSettingsManager {
.setInitial(Durations.fromMillis(config.getGrpcClientProducerBackoffInitialMillis()))
.setMax(Durations.fromMillis(config.getGrpcClientProducerBackoffMaxMillis()))
.setMultiplier(config.getGrpcClientProducerBackoffMultiplier())
.build())
.build();
.build());
builder.getPublishingBuilder()
.setValidateMessageType(config.isEnableTopicMessageTypeCheck())
@@ -22,6 +22,10 @@ import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
import java.net.URL;
import org.apache.rocketmq.client.log.ClientLogger;
import org.assertj.core.util.Strings;
import java.io.IOException;
import java.io.InputStream;
import org.junit.After;
import org.junit.Before;
import org.slf4j.LoggerFactory;
@@ -37,7 +41,11 @@ public class InitConfigAndLoggerTest {
if (mockProxyHomeURL != null) {
mockProxyHome = mockProxyHomeURL.toURI().getPath();
}
System.setProperty(RMQ_PROXY_HOME, mockProxyHome);
if (!Strings.isNullOrEmpty(mockProxyHome)) {
System.setProperty(RMQ_PROXY_HOME, mockProxyHome);
}
ConfigurationManager.initEnv();
ConfigurationManager.intConfig();
initLogger();
@@ -56,8 +64,18 @@ public class InitConfigAndLoggerTest {
JoranConfigurator configurator = new JoranConfigurator();
configurator.setContext(lc);
lc.reset();
//https://logback.qos.ch/manual/configuration.html
// https://logback.qos.ch/manual/configuration.html
lc.setPackagingDataEnabled(false);
try (InputStream inputStream = InitConfigAndLoggerTest.class.getClassLoader()
.getResourceAsStream("rmq-proxy-home/conf/logback_proxy.xml")) {
if (null != inputStream) {
configurator.doConfigure(inputStream);
return;
}
} catch (IOException ignore) {
}
configurator.doConfigure(mockProxyHome + "/conf/logback_proxy.xml");
}
}
@@ -34,6 +34,7 @@ import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.InitConfigAndLoggerTest;
import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -77,9 +78,10 @@ public class GrpcMessagingApplicationTest extends InitConfigAndLoggerTest {
metadata.put(InterceptorConstants.LANGUAGE, JAVA);
metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR);
metadata.put(InterceptorConstants.LOCAL_ADDRESS, LOCAL_ADDR);
Context.current()
.withValue(InterceptorConstants.METADATA, metadata)
.attach();
Assert.assertNotNull(Context.current()
.withValue(InterceptorConstants.METADATA, metadata)
.attach());
CompletableFuture<QueryRouteResponse> future = new CompletableFuture<>();
QueryRouteRequest request = QueryRouteRequest.newBuilder()
@@ -105,9 +107,10 @@ public class GrpcMessagingApplicationTest extends InitConfigAndLoggerTest {
metadata.put(InterceptorConstants.LANGUAGE, JAVA);
metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR);
metadata.put(InterceptorConstants.LOCAL_ADDRESS, LOCAL_ADDR);
Context.current()
.withValue(InterceptorConstants.METADATA, metadata)
.attach();
Assert.assertNotNull(Context.current()
.withValue(InterceptorConstants.METADATA, metadata)
.attach());
QueryRouteRequest request = QueryRouteRequest.newBuilder()
.setEndpoints(grpcEndpoints)
+50
View File
@@ -0,0 +1,50 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "remoting",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//logging",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":remoting",
"//:test_deps",
"@maven//:io_netty_netty_all",
"@maven//:com_google_code_gson_gson",
"@maven//:com_alibaba_fastjson",
],
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",
],
)
@@ -175,8 +175,7 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
private static int initValueIndex() {
Random r = new Random();
return Math.abs(r.nextInt() % 999) % 999;
return r.nextInt(999);
}
@Override
@@ -80,13 +80,11 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
private DefaultEventExecutorGroup defaultEventExecutorGroup;
/**
* NettyRemotingServer may holds multiple SubRemotingServer, each server will be stored in this container with a
* NettyRemotingServer may hold multiple SubRemotingServer, each server will be stored in this container with a
* ListenPort key.
*/
private ConcurrentMap<Integer/*Port*/, NettyRemotingAbstract> remotingServerTable = new ConcurrentHashMap<Integer, NettyRemotingAbstract>();
private int port = 0;
private static final String HANDSHAKE_HANDLER_NAME = "handshakeHandler";
private static final String TLS_HANDLER_NAME = "sslHandler";
private static final String FILE_REGION_ENCODER_NAME = "fileRegionEncoder";
@@ -163,8 +161,6 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
}
loadSslContext();
this.remotingServerTable.put(this.nettyServerConfig.getListenPort(), this);
}
public void loadSslContext() {
@@ -247,9 +243,13 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
}
try {
ChannelFuture sync = this.serverBootstrap.bind().sync();
ChannelFuture sync = this.serverBootstrap.bind(nettyServerConfig.getListenPort()).sync();
InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
this.port = addr.getPort();
if (0 == nettyServerConfig.getListenPort()) {
this.nettyServerConfig.setListenPort(addr.getPort());
log.debug("Server is listening {}", this.nettyServerConfig.getListenPort());
}
this.remotingServerTable.put(this.nettyServerConfig.getListenPort(), this);
} catch (InterruptedException e1) {
throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
}
@@ -320,7 +320,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
@Override
public int localListenPort() {
return this.port;
return this.nettyServerConfig.getListenPort();
}
@Override
@@ -540,7 +540,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
* resources from its parent server.
*/
class SubRemotingServer extends NettyRemotingAbstract implements RemotingServer {
private final int listenPort;
private volatile int listenPort;
private volatile Channel serverChannel;
SubRemotingServer(final int port, final int permitsOnway, final int permitsAsync) {
@@ -613,7 +613,14 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
@Override
public void start() {
try {
if (listenPort < 0) {
listenPort = 0;
}
this.serverChannel = NettyRemotingServer.this.serverBootstrap.bind(listenPort).sync().channel();
if (0 == listenPort) {
InetSocketAddress addr = (InetSocketAddress) this.serverChannel.localAddress();
this.listenPort = addr.getPort();
}
} catch (InterruptedException e) {
throw new RuntimeException("this.subRemotingServer.serverBootstrap.bind().sync() InterruptedException", e);
}
@@ -17,7 +17,7 @@
package org.apache.rocketmq.remoting.netty;
public class NettyServerConfig implements Cloneable {
private int listenPort = 8888;
private int listenPort = 0;
private int serverWorkerThreads = 8;
private int serverCallbackExecutorThreads = 0;
private int serverSelectorThreads = 3;
@@ -34,6 +34,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class RemotingServerTest {
@@ -90,8 +91,8 @@ public class RemotingServerTest {
requestHeader.setCount(1);
requestHeader.setMessageTitle("Welcome");
RemotingCommand request = RemotingCommand.createRequestCommand(0, requestHeader);
RemotingCommand response = remotingClient.invokeSync("localhost:8888", request, 1000 * 3);
assertTrue(response != null);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), request, 1000 * 3);
assertNotNull(response);
assertThat(response.getLanguage()).isEqualTo(LanguageCode.JAVA);
assertThat(response.getExtFields()).hasSize(2);
@@ -103,7 +104,7 @@ public class RemotingServerTest {
RemotingCommand request = RemotingCommand.createRequestCommand(0, null);
request.setRemark("messi");
remotingClient.invokeOneway("localhost:8888", request, 1000 * 3);
remotingClient.invokeOneway("localhost:" + remotingServer.localListenPort(), request, 1000 * 3);
}
@Test
@@ -113,7 +114,7 @@ public class RemotingServerTest {
final CountDownLatch latch = new CountDownLatch(1);
RemotingCommand request = RemotingCommand.createRequestCommand(0, null);
request.setRemark("messi");
remotingClient.invokeAsync("localhost:8888", request, 1000 * 3, new InvokeCallback() {
remotingClient.invokeAsync("localhost:" + remotingServer.localListenPort(), request, 1000 * 3, new InvokeCallback() {
@Override
public void operationComplete(ResponseFuture responseFuture) {
latch.countDown();
@@ -57,7 +57,7 @@ public class SubRemotingServerTest {
subServer.registerProcessor(1, new NettyRequestProcessor() {
@Override
public RemotingCommand processRequest(final ChannelHandlerContext ctx,
final RemotingCommand request) throws Exception {
final RemotingCommand request) throws Exception {
request.setRemark(String.valueOf(RemotingHelper.parseSocketAddressPort(ctx.channel().localAddress())));
return request;
}
@@ -72,14 +72,16 @@ public class SubRemotingServerTest {
}
@Test
public void testInvokeSubRemotingServer() throws InterruptedException, RemotingTimeoutException, RemotingConnectException, RemotingSendRequestException {
public void testInvokeSubRemotingServer() throws InterruptedException, RemotingTimeoutException,
RemotingConnectException, RemotingSendRequestException {
RequestHeader requestHeader = new RequestHeader();
requestHeader.setCount(1);
requestHeader.setMessageTitle("Welcome");
// Parent remoting server doesn't support RequestCode 1
RemotingCommand request = RemotingCommand.createRequestCommand(1, requestHeader);
RemotingCommand response = remotingClient.invokeSync("localhost:8888", request, 1000 * 3);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), request,
1000 * 3);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED);
@@ -19,9 +19,14 @@ package org.apache.rocketmq.remoting;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.UUID;
import java.io.InputStream;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import org.apache.rocketmq.remoting.common.TlsMode;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
@@ -65,7 +70,7 @@ import static org.apache.rocketmq.remoting.netty.TlsSystemConfig.tlsServerTrustC
import static org.apache.rocketmq.remoting.netty.TlsSystemConfig.tlsTestModeEnable;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertNotNull;
@RunWith(MockitoJUnitRunner.class)
public class TlsTest {
@@ -198,7 +203,7 @@ public class TlsTest {
@Test
public void serverRejectsSSLClient() throws Exception {
try {
RemotingCommand response = remotingClient.invokeSync("localhost:8888", createRequest(), 1000 * 5);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 5);
failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
} catch (RemotingSendRequestException ignore) {
}
@@ -211,7 +216,7 @@ public class TlsTest {
@Test
public void serverRejectsUntrustedClientCert() throws Exception {
try {
RemotingCommand response = remotingClient.invokeSync("localhost:8888", createRequest(), 1000 * 5);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 5);
failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
} catch (RemotingSendRequestException ignore) {
}
@@ -229,7 +234,7 @@ public class TlsTest {
@Test
public void noClientAuthFailure() throws Exception {
try {
RemotingCommand response = remotingClient.invokeSync("localhost:8888", createRequest(), 1000 * 3);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 3);
failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
} catch (RemotingSendRequestException ignore) {
}
@@ -242,7 +247,7 @@ public class TlsTest {
@Test
public void clientRejectsUntrustedServerCert() throws Exception {
try {
RemotingCommand response = remotingClient.invokeSync("localhost:8888", createRequest(), 1000 * 3);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 3);
failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
} catch (RemotingSendRequestException ignore) {
}
@@ -301,8 +306,31 @@ public class TlsTest {
}
private static String getCertsPath(String fileName) {
File resourcesDirectory = new File("src/test/resources/certs");
return resourcesDirectory.getAbsolutePath() + "/" + fileName;
ClassLoader loader = TlsTest.class.getClassLoader();
InputStream stream = loader.getResourceAsStream("certs/" + fileName);
if (null == stream) {
throw new RuntimeException("File: " + fileName + " is not found");
}
try {
String[] segments = fileName.split("\\.");
File f = File.createTempFile(UUID.randomUUID().toString(), segments[1]);
f.deleteOnExit();
try (BufferedInputStream bis = new BufferedInputStream(stream);
BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(f))) {
byte[] buffer = new byte[1024];
int len;
while ((len = bis.read(buffer)) > 0) {
bos.write(buffer, 0, len);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
return f.getAbsolutePath();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
private static RemotingCommand createRequest() {
@@ -317,8 +345,8 @@ public class TlsTest {
}
private void requestThenAssertResponse(RemotingClient remotingClient) throws Exception {
RemotingCommand response = remotingClient.invokeSync("localhost:8888", createRequest(), 1000 * 3);
assertTrue(response != null);
RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 3);
assertNotNull(response);
assertThat(response.getLanguage()).isEqualTo(LanguageCode.JAVA);
assertThat(response.getExtFields()).hasSize(2);
assertThat(response.getExtFields().get("messageTitle")).isEqualTo("Welcome");
+59
View File
@@ -0,0 +1,59 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "srvutil",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//common",
"//remoting",
"//logging",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:commons_cli_commons_cli",
"@maven//:com_googlecode_concurrentlinkedhashmap_concurrentlinkedhashmap_lru",
"@maven//:com_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":srvutil",
"//common",
"//remoting",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
],
)
GenTestRules(
name = "GeneratedTestRules",
test_files = glob(["src/test/java/**/*Test.java"]),
deps = [
":tests",
],
)
+80
View File
@@ -0,0 +1,80 @@
#
# 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.
#
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",
"//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_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
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",
],
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",
"src/test/java/org/apache/rocketmq/store/MappedFileQueueTest",
"src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest",
],
medium_tests = [
"src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest",
"src/test/java/org/apache/rocketmq/store/HATest",
],
)
Binary file not shown.
@@ -707,6 +707,10 @@ public class MessageStoreConfig {
}
public void setHaListenPort(int haListenPort) {
if (haListenPort < 0) {
this.haListenPort = 0;
return;
}
this.haListenPort = haListenPort;
}
@@ -38,6 +38,7 @@ import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
public class DefaultHAService implements HAService {
@@ -66,8 +67,7 @@ public class DefaultHAService implements HAService {
@Override
public void init(final DefaultMessageStore defaultMessageStore) throws IOException {
this.defaultMessageStore = defaultMessageStore;
this.acceptSocketService =
new DefaultAcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
this.acceptSocketService = new DefaultAcceptSocketService(defaultMessageStore.getMessageStoreConfig());
this.groupTransferService = new GroupTransferService(this, defaultMessageStore);
if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE) {
this.haClient = new DefaultHAClient(this.defaultMessageStore);
@@ -258,8 +258,8 @@ public class DefaultHAService implements HAService {
class DefaultAcceptSocketService extends AcceptSocketService {
public DefaultAcceptSocketService(int port) {
super(port);
public DefaultAcceptSocketService(final MessageStoreConfig messageStoreConfig) {
super(messageStoreConfig);
}
@Override
@@ -284,8 +284,11 @@ public class DefaultHAService implements HAService {
private ServerSocketChannel serverSocketChannel;
private Selector selector;
public AcceptSocketService(final int port) {
this.socketAddressListen = new InetSocketAddress(port);
private final MessageStoreConfig messageStoreConfig;
public AcceptSocketService(final MessageStoreConfig messageStoreConfig) {
this.messageStoreConfig = messageStoreConfig;
this.socketAddressListen = new InetSocketAddress(messageStoreConfig.getHaListenPort());
}
/**
@@ -298,6 +301,10 @@ public class DefaultHAService implements HAService {
this.selector = RemotingUtil.openSelector();
this.serverSocketChannel.socket().setReuseAddress(true);
this.serverSocketChannel.socket().bind(this.socketAddressListen);
if (0 == messageStoreConfig.getHaListenPort()) {
messageStoreConfig.setHaListenPort(this.serverSocketChannel.socket().getLocalPort());
log.info("OS picked up {} to listen for HA", messageStoreConfig.getHaListenPort());
}
this.serverSocketChannel.configureBlocking(false);
this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
}
@@ -309,8 +316,13 @@ public class DefaultHAService implements HAService {
public void shutdown(final boolean interrupt) {
super.shutdown(interrupt);
try {
this.serverSocketChannel.close();
this.selector.close();
if (null != this.serverSocketChannel) {
this.serverSocketChannel.close();
}
if (null != this.selector) {
this.selector.close();
}
} catch (IOException e) {
log.error("AcceptSocketService shutdown exception", e);
}
@@ -39,6 +39,7 @@ import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.ha.DefaultHAService;
import org.apache.rocketmq.store.ha.GroupTransferService;
import org.apache.rocketmq.store.ha.HAClient;
@@ -69,7 +70,7 @@ public class AutoSwitchHAService extends DefaultHAService {
this.epochCache = new EpochFileCache(defaultMessageStore.getMessageStoreConfig().getStorePathEpochFile());
this.epochCache.initCacheFromFile();
this.defaultMessageStore = defaultMessageStore;
this.acceptSocketService = new AutoSwitchAcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
this.acceptSocketService = new AutoSwitchAcceptSocketService(defaultMessageStore.getMessageStoreConfig());
this.groupTransferService = new GroupTransferService(this, defaultMessageStore);
this.haConnectionStateNotificationService = new HAConnectionStateNotificationService(this, defaultMessageStore);
}
@@ -403,8 +404,8 @@ public class AutoSwitchHAService extends DefaultHAService {
class AutoSwitchAcceptSocketService extends AcceptSocketService {
public AutoSwitchAcceptSocketService(int port) {
super(port);
public AutoSwitchAcceptSocketService(final MessageStoreConfig messageStoreConfig) {
super(messageStoreConfig);
}
@Override
@@ -53,8 +53,8 @@ public class AppendCallbackTest {
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 4);
messageStoreConfig.setMaxHashSlotNum(100);
messageStoreConfig.setMaxIndexNum(100 * 10);
messageStoreConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "unitteststore");
messageStoreConfig.setStorePathCommitLog(System.getProperty("user.home") + File.separator + "unitteststore" + File.separator + "commitlog");
messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore");
messageStoreConfig.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore" + File.separator + "commitlog");
//too much reference
DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, new BrokerConfig());
CommitLog commitLog = new CommitLog(messageStore);
@@ -63,7 +63,7 @@ public class AppendCallbackTest {
@After
public void destroy() {
UtilAll.deleteFile(new File(System.getProperty("user.home") + File.separator + "unitteststore"));
UtilAll.deleteFile(new File(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore"));
}
@Test
@@ -64,7 +64,7 @@ public class BatchPutMessageTest {
messageStore.shutdown();
messageStore.destroy();
UtilAll.deleteFile(new File(System.getProperty("user.home") + File.separator + "putmessagesteststore"));
UtilAll.deleteFile(new File(System.getProperty("java.io.tmpdir") + File.separator + "putmessagesteststore"));
}
private MessageStore buildMessageStore() throws Exception {
@@ -75,8 +75,10 @@ public class BatchPutMessageTest {
messageStoreConfig.setMaxIndexNum(100 * 10);
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setFlushIntervalConsumeQueue(1);
messageStoreConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "putmessagesteststore");
messageStoreConfig.setStorePathCommitLog(System.getProperty("user.home") + File.separator + "putmessagesteststore" + File.separator + "commitlog");
messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "putmessagesteststore");
messageStoreConfig.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + File.separator
+ "putmessagesteststore" + File.separator + "commitlog");
messageStoreConfig.setHaListenPort(0);
return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), new MyMessageArrivingListener(), new BrokerConfig());
}
@@ -46,7 +46,7 @@ public class ConsumeQueueTest {
private static final String topic = "abc";
private static final int queueId = 0;
private static final String storePath = "." + File.separator + "unit_test_store";
private static final String storePath = System.getProperty("java.io.tmpdir") + File.separator + "unit_test_store";
private static final int commitLogFileSize = 1024 * 8;
private static final int cqFileSize = 10 * 20;
private static final int cqExtFileSize = 10 * (ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE + 64);
@@ -119,7 +119,7 @@ public class ConsumeQueueTest {
messageStoreConfig.setMappedFileSizeConsumeQueueExt(cqExtFileSize);
messageStoreConfig.setMessageIndexEnable(false);
messageStoreConfig.setEnableConsumeQueueExt(enableCqExt);
messageStoreConfig.setHaListenPort(0);
messageStoreConfig.setStorePathRootDir(storePath);
messageStoreConfig.setStorePathCommitLog(storePath + File.separator + "commitlog");
@@ -473,7 +473,7 @@ public class DefaultMessageStoreCleanFilesTest {
messageStoreConfig.setDeleteWhen(deleteWhen);
messageStoreConfig.setDiskMaxUsedSpaceRatio(diskMaxUsedSpaceRatio);
String storePathRootDir = System.getProperty("user.home") + File.separator
String storePathRootDir = System.getProperty("java.io.tmpdir") + File.separator
+ "DefaultMessageStoreCleanFilesTest-" + UUID.randomUUID();
String storePathCommitLog = storePathRootDir + File.separator + "commitlog";
messageStoreConfig.setStorePathRootDir(storePathRootDir);
@@ -70,7 +70,10 @@ public class DefaultMessageStoreShutDownTest {
messageStoreConfig.setMaxHashSlotNum(10000);
messageStoreConfig.setMaxIndexNum(100 * 100);
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setHaListenPort(StoreTestBase.nextPort());
messageStoreConfig.setHaListenPort(0);
String storeRootPath = System.getProperty("java.io.tmpdir") + File.separator + "store";
messageStoreConfig.setStorePathRootDir(storeRootPath);
messageStoreConfig.setHaListenPort(0);
return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), null, new BrokerConfig());
}
@@ -33,6 +33,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.common.BrokerConfig;
@@ -49,6 +50,7 @@ import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.assertj.core.util.Strings;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -90,6 +92,8 @@ public class DefaultMessageStoreTest {
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 4);
messageStoreConfig.setMaxHashSlotNum(100);
messageStoreConfig.setMaxIndexNum(100 * 10);
messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "store");
messageStoreConfig.setHaListenPort(0);
MessageStore master = new DefaultMessageStore(messageStoreConfig, null, new MyMessageArrivingListener(), new BrokerConfig());
boolean load = master.load();
@@ -115,6 +119,10 @@ public class DefaultMessageStoreTest {
}
private MessageStore buildMessageStore() throws Exception {
return buildMessageStore(null);
}
private MessageStore buildMessageStore(String storePathRootDir) throws Exception {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setMappedFileSizeCommitLog(1024 * 1024 * 10);
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 1024 * 10);
@@ -122,8 +130,16 @@ public class DefaultMessageStoreTest {
messageStoreConfig.setMaxIndexNum(100 * 100);
messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
messageStoreConfig.setFlushIntervalConsumeQueue(1);
messageStoreConfig.setHaListenPort(StoreTestBase.nextPort());
return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest", true), new MyMessageArrivingListener(), new BrokerConfig());
messageStoreConfig.setHaListenPort(0);
if (Strings.isNullOrEmpty(storePathRootDir)) {
UUID uuid = UUID.randomUUID();
storePathRootDir = System.getProperty("java.io.tmpdir") + File.separator + "store-" + uuid.toString();
}
messageStoreConfig.setStorePathRootDir(storePathRootDir);
return new DefaultMessageStore(messageStoreConfig,
new BrokerStatsManager("simpleTest", true),
new MyMessageArrivingListener(),
new BrokerConfig());
}
@Test
@@ -592,14 +608,15 @@ public class DefaultMessageStoreTest {
//1.just reboot
messageStore.shutdown();
messageStore = buildMessageStore();
String storeRootDir = ((DefaultMessageStore) messageStore).getMessageStoreConfig().getStorePathRootDir();
messageStore = buildMessageStore(storeRootDir);
boolean load = messageStore.load();
assertTrue(load);
messageStore.start();
assertTrue(maxPhyOffset == messageStore.getMaxPhyOffset());
assertTrue(maxCqOffset == messageStore.getMaxOffsetInQueue(topic, 0));
//2.damage commitlog and reboot normal
//2.damage commit-log and reboot normal
for (int i = 0; i < 100; i++) {
MessageExtBrokerInner messageExtBrokerInner = buildMessage();
messageExtBrokerInner.setTopic(topic);
@@ -619,10 +636,10 @@ public class DefaultMessageStoreTest {
messageStore.shutdown();
//damage last message
damageCommitlog(secondLastPhyOffset);
damageCommitLog((DefaultMessageStore) messageStore, secondLastPhyOffset);
//reboot
messageStore = buildMessageStore();
messageStore = buildMessageStore(storeRootDir);
load = messageStore.load();
assertTrue(load);
messageStore.start();
@@ -648,14 +665,14 @@ public class DefaultMessageStoreTest {
messageStore.shutdown();
//damage last message
damageCommitlog(secondLastPhyOffset);
damageCommitLog((DefaultMessageStore) messageStore, secondLastPhyOffset);
//add abort file
String fileName = StorePathConfigHelper.getAbortFile(((DefaultMessageStore) messageStore).getMessageStoreConfig().getStorePathRootDir());
File file = new File(fileName);
UtilAll.ensureDirOK(file.getParent());
file.createNewFile();
messageStore = buildMessageStore();
messageStore = buildMessageStore(storeRootDir);
load = messageStore.load();
assertTrue(load);
messageStore.start();
@@ -687,24 +704,23 @@ public class DefaultMessageStoreTest {
return false;
}
private void damageCommitlog(long offset) throws Exception {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
private void damageCommitLog(DefaultMessageStore store, long offset) throws Exception {
assertThat(store).isNotNull();
MessageStoreConfig messageStoreConfig = store.getMessageStoreConfig();
File file = new File(messageStoreConfig.getStorePathCommitLog() + File.separator + "00000000000000000000");
FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 1024 * 1024 * 10);
int bodyLen = mappedByteBuffer.getInt((int) offset + 84);
int topicLenIndex = (int) offset + 84 + bodyLen + 4;
mappedByteBuffer.position(topicLenIndex);
mappedByteBuffer.putInt(0);
mappedByteBuffer.putInt(0);
mappedByteBuffer.putInt(0);
mappedByteBuffer.putInt(0);
mappedByteBuffer.force();
fileChannel.force(true);
fileChannel.close();
try (RandomAccessFile raf = new RandomAccessFile(file, "rw");
FileChannel fileChannel = raf.getChannel()) {
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 1024 * 1024 * 10);
int bodyLen = mappedByteBuffer.getInt((int) offset + 84);
int topicLenIndex = (int) offset + 84 + bodyLen + 4;
mappedByteBuffer.position(topicLenIndex);
mappedByteBuffer.putInt(0);
mappedByteBuffer.putInt(0);
mappedByteBuffer.putInt(0);
mappedByteBuffer.putInt(0);
mappedByteBuffer.force();
fileChannel.force(true);
}
}
@Test
@@ -33,7 +33,6 @@ import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
@@ -41,10 +40,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.*;
/**
* HATest
*
*/
public class HATest {
private final String StoreMessage = "Once, there was a chance for me!";
private int QUEUE_TOTAL = 100;
@@ -58,7 +53,7 @@ public class HATest {
private MessageStoreConfig masterMessageStoreConfig;
private MessageStoreConfig slaveStoreConfig;
private BrokerStatsManager brokerStatsManager = new BrokerStatsManager("simpleTest", true);
private String storePathRootParentDir = System.getProperty("user.home") + File.separator +
private String storePathRootParentDir = System.getProperty("java.io.tmpdir") + File.separator +
UUID.randomUUID().toString().replace("-", "");
private String storePathRootDir = storePathRootParentDir + File.separator + "store";
@@ -70,6 +65,7 @@ public class HATest {
masterMessageStoreConfig.setBrokerRole(BrokerRole.SYNC_MASTER);
masterMessageStoreConfig.setStorePathRootDir(storePathRootDir + File.separator + "master");
masterMessageStoreConfig.setStorePathCommitLog(storePathRootDir + File.separator + "master" + File.separator + "commitlog");
masterMessageStoreConfig.setHaListenPort(0);
masterMessageStoreConfig.setTotalReplicas(2);
masterMessageStoreConfig.setInSyncReplicas(2);
buildMessageStoreConfig(masterMessageStoreConfig);
@@ -77,7 +73,7 @@ public class HATest {
slaveStoreConfig.setBrokerRole(BrokerRole.SLAVE);
slaveStoreConfig.setStorePathRootDir(storePathRootDir + File.separator + "slave");
slaveStoreConfig.setStorePathCommitLog(storePathRootDir + File.separator + "slave" + File.separator + "commitlog");
slaveStoreConfig.setHaListenPort(10943);
slaveStoreConfig.setHaListenPort(0);
slaveStoreConfig.setTotalReplicas(2);
slaveStoreConfig.setInSyncReplicas(2);
buildMessageStoreConfig(slaveStoreConfig);
@@ -88,8 +84,10 @@ public class HATest {
assertTrue(load);
assertTrue(slaveLoad);
messageStore.start();
slaveMessageStore.updateHaMasterAddress("127.0.0.1:" + masterMessageStoreConfig.getHaListenPort());
slaveMessageStore.start();
slaveMessageStore.updateHaMasterAddress("127.0.0.1:10912");
slaveMessageStore.updateHaMasterAddress("127.0.0.1:" + masterMessageStoreConfig.getHaListenPort());
Thread.sleep(6000L);//because the haClient will wait 5s after the first connectMaster failed,sleep 6s
}
@@ -119,7 +117,7 @@ public class HATest {
for (long i = 0; i < totalMsgs; i++) {
GetMessageResult result = slaveMessageStore.getMessage("GROUP_A", "FooBar", 0, i, 1024 * 1024, null);
assertThat(result).isNotNull();
assertTrue(GetMessageStatus.FOUND.equals(result.getStatus()));
assertEquals(GetMessageStatus.FOUND, result.getStatus());
result.release();
}
}
@@ -138,7 +136,7 @@ public class HATest {
//so direct read from commitLog by physical offset
MessageExt slaveMsg = slaveMessageStore.lookMessageByOffset(result.getAppendMessageResult().getWroteOffset());
assertNotNull(slaveMsg);
assertTrue(Arrays.equals(msg.getBody(), slaveMsg.getBody()));
assertArrayEquals(msg.getBody(), slaveMsg.getBody());
assertEquals(msg.getTopic(), slaveMsg.getTopic());
assertEquals(msg.getTags(), slaveMsg.getTags());
assertEquals(msg.getKeys(), slaveMsg.getKeys());
@@ -171,7 +169,7 @@ public class HATest {
//so direct read from commitLog by physical offset
MessageExt slaveMsg = slaveMessageStore.lookMessageByOffset(result.getAppendMessageResult().getWroteOffset());
assertNotNull(slaveMsg);
assertTrue(Arrays.equals(msg.getBody(), slaveMsg.getBody()));
assertArrayEquals(msg.getBody(), slaveMsg.getBody());
assertEquals(msg.getTopic(), slaveMsg.getTopic());
assertEquals(msg.getTags(), slaveMsg.getTags());
assertEquals(msg.getKeys(), slaveMsg.getKeys());
@@ -208,7 +206,7 @@ public class HATest {
//so direct read from commitLog by physical offset
MessageExt slaveMsg = slaveMessageStore.lookMessageByOffset(result.getAppendMessageResult().getWroteOffset());
assertNotNull(slaveMsg);
assertTrue(Arrays.equals(msg.getBody(), slaveMsg.getBody()));
assertArrayEquals(msg.getBody(), slaveMsg.getBody());
assertEquals(msg.getTopic(), slaveMsg.getTopic());
assertEquals(msg.getTags(), slaveMsg.getTags());
assertEquals(msg.getKeys(), slaveMsg.getKeys());
@@ -49,9 +49,9 @@ public class MultiDispatchTest {
messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 4);
messageStoreConfig.setMaxHashSlotNum(100);
messageStoreConfig.setMaxIndexNum(100 * 10);
messageStoreConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "unitteststore1");
messageStoreConfig.setStorePathRootDir(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore1");
messageStoreConfig.setStorePathCommitLog(
System.getProperty("user.home") + File.separator + "unitteststore1" + File.separator + "commitlog");
System.getProperty("java.io.tmpdir") + File.separator + "unitteststore1" + File.separator + "commitlog");
messageStoreConfig.setEnableLmq(true);
messageStoreConfig.setEnableMultiDispatch(true);
@@ -64,7 +64,7 @@ public class MultiDispatchTest {
@After
public void destroy() {
UtilAll.deleteFile(new File(System.getProperty("user.home") + File.separator + "unitteststore1"));
UtilAll.deleteFile(new File(System.getProperty("java.io.tmpdir") + File.separator + "unitteststore1"));
}
@Test
@@ -150,7 +150,7 @@ public class StoreTestBase {
}
public static String createBaseDir() {
String baseDir = System.getProperty("user.home") + File.separator + "unitteststore" + File.separator + UUID.randomUUID();
String baseDir = System.getProperty("java.io.tmpdir") + File.separator + "unitteststore" + File.separator + UUID.randomUUID();
final File file = new File(baseDir);
if (file.exists()) {
System.exit(1);
@@ -21,7 +21,7 @@ import java.util.UUID;
public class StoreTestUtils {
public static String createBaseDir() {
String baseDir = System.getProperty("user.home") + File.separator + "unitteststore-" + UUID.randomUUID();
String baseDir = System.getProperty("java.io.tmpdir") + File.separator + "unitteststore-" + UUID.randomUUID();
final File file = new File(baseDir);
if (file.exists()) {
System.exit(1);
@@ -100,8 +100,6 @@ public class TimerMessageStoreTest {
messageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("TimerTest",false), new MyMessageArrivingListener(), new BrokerConfig());
boolean load = messageStore.load();
List<PutMessageHook> putMessageHookList = messageStore.getPutMessageHookList();
assertTrue(load);
messageStore.start();
}
+107
View File
@@ -0,0 +1,107 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "test",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//broker",
"//client",
"//common",
"//remoting",
"//logging",
"//srvutil",
"//tools",
"//namesrv",
"//controller",
"//container",
"//proxy",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:log4j_log4j",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:ch_qos_logback_logback_core",
"@maven//:com_google_truth_truth",
"@maven//:javax_annotation_javax_annotation_api",
"@maven//:org_awaitility_awaitility",
"@maven//:org_reflections_reflections",
"@maven//:com_google_protobuf_protobuf_java_util",
"@maven//:commons_cli_commons_cli",
"@maven//:com_google_guava_guava",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":test",
"//broker",
"//client",
"//common",
"//remoting",
"//logging",
"//tools",
"//:test_deps",
"//store",
"//namesrv",
"//controller",
"//container",
"//proxy",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:com_google_truth_truth",
"@maven//:log4j_log4j",
"@maven//:io_grpc_grpc_testing",
"@maven//:com_google_protobuf_protobuf_java_util",
"@maven//:com_google_protobuf_protobuf_java",
"@maven//:io_grpc_grpc_netty_shaded",
"@maven//:io_grpc_grpc_context",
"@maven//:io_grpc_grpc_stub",
"@maven//:io_grpc_grpc_api",
"@maven//:org_apache_rocketmq_rocketmq_proto",
"@maven//:org_slf4j_slf4j_api",
"@maven//:com_google_guava_guava",
],
resources = [
"src/test/resources/rmq-proxy-home/conf/broker.conf",
"src/test/resources/rmq-proxy-home/conf/logback_proxy.xml",
"src/test/resources/rmq-proxy-home/conf/rmq-proxy.json",
"src/test/resources/log4j.xml",
"src/test/resources/logback-test.xml",
] + glob(["src/test/resources/schema/**/*.schema"]),
)
GenTestRules(
name = "GeneratedTestRules",
default_test_size = "large",
test_files = glob(["src/test/java/**/*IT.java"]),
deps = [
":tests",
],
exclude_tests = [
"src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseIT",
]
)
@@ -177,7 +177,7 @@ public class SchemaTools {
}
});
String key = String.format("Method %s(%s)", method.getName(), Arrays.stream(parameterTypes).map(Class::getName).collect(Collectors.joining(",")));
String value = String.format("%s throws (%s)",
String value = String.format("%s throws (%s): %s",
isPublicOrPrivate(method.getModifiers()),
method.getReturnType().getName(),
Arrays.stream(exceptionTypes).map(Class::getName).collect(Collectors.joining(",")));
@@ -265,7 +265,7 @@ public final class RandomUtil {
Random rd = new Random();
int index = 0;
for (int i = 0; i < result.length; i++) {
index = Math.abs(rd.nextInt() % len--);
index = rd.nextInt(len--);
result[i] = source[index];
source[index] = source[len];
}
@@ -56,15 +56,12 @@ public class IntegrationTestBase {
public static volatile int COMMIT_LOG_SIZE = 1024 * 1024 * 100;
protected static final int INDEX_NUM = 1000;
private static final AtomicInteger port = new AtomicInteger(40000);
public static synchronized int nextPort() {
return port.addAndGet(random.nextInt(10) + 10);
}
protected static Random random = new Random();
static {
System.setProperty("rocketmq.client.logRoot", System.getProperty("java.io.tmpdir"));
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
@@ -99,7 +96,7 @@ public class IntegrationTestBase {
}
public static String createBaseDir() {
String baseDir = System.getProperty("user.home") + SEP + "unitteststore-" + UUID.randomUUID();
String baseDir = System.getProperty("java.io.tmpdir") + SEP + "unitteststore-" + UUID.randomUUID();
final File file = new File(baseDir);
if (file.exists()) {
logger.info(String.format("[%s] has already existed, please back up and remove it for integration tests", baseDir));
@@ -116,7 +113,7 @@ public class IntegrationTestBase {
namesrvConfig.setKvConfigPath(baseDir + SEP + "namesrv" + SEP + "kvConfig.json");
namesrvConfig.setConfigStorePath(baseDir + SEP + "namesrv" + SEP + "namesrv.properties");
nameServerNettyServerConfig.setListenPort(nextPort());
nameServerNettyServerConfig.setListenPort(0);
NamesrvController namesrvController = new NamesrvController(namesrvConfig, nameServerNettyServerConfig);
try {
Truth.assertThat(namesrvController.initialize()).isTrue();
@@ -154,8 +151,8 @@ public class IntegrationTestBase {
public static BrokerController createAndStartBroker(MessageStoreConfig storeConfig, BrokerConfig brokerConfig) {
NettyServerConfig nettyServerConfig = new NettyServerConfig();
NettyClientConfig nettyClientConfig = new NettyClientConfig();
nettyServerConfig.setListenPort(nextPort());
storeConfig.setHaListenPort(nextPort());
nettyServerConfig.setListenPort(0);
storeConfig.setHaListenPort(0);
BrokerController brokerController = new BrokerController(brokerConfig, nettyServerConfig, nettyClientConfig, storeConfig);
try {
Truth.assertThat(brokerController.initialize()).isTrue();
@@ -38,7 +38,6 @@ import org.apache.rocketmq.test.factory.ProducerFactory;
import org.junit.Assert;
import org.junit.Test;
import static org.apache.rocketmq.test.base.IntegrationTestBase.nextPort;
import static sun.util.locale.BaseLocale.SEP;
public class DLedgerProduceAndConsumeIT {
@@ -57,7 +56,7 @@ public class DLedgerProduceAndConsumeIT {
String baseDir = IntegrationTestBase.createBaseDir();
storeConfig.setStorePathRootDir(baseDir);
storeConfig.setStorePathCommitLog(baseDir + SEP + "commitlog");
storeConfig.setHaListenPort(nextPort());
storeConfig.setHaListenPort(0);
storeConfig.setMappedFileSizeCommitLog(10 * 1024 * 1024);
storeConfig.setEnableDLegerCommitLog(true);
storeConfig.setdLegerGroup(brokerName);
@@ -71,7 +70,8 @@ public class DLedgerProduceAndConsumeIT {
String cluster = UUID.randomUUID().toString();
String brokerName = UUID.randomUUID().toString();
String selfId = "n0";
String peers = String.format("n0-localhost:%d", nextPort());
// TODO: We need to acquire the actual listening port after the peer has started.
String peers = String.format("n0-localhost:%d", 0);
BrokerConfig brokerConfig = buildBrokerConfig(cluster, brokerName);
MessageStoreConfig storeConfig = buildStoreConfig(brokerName, peers, selfId);
BrokerController brokerController = IntegrationTestBase.createAndStartBroker(storeConfig, brokerConfig);
@@ -23,21 +23,21 @@ import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.factory.ConsumerFactory;
import org.apache.rocketmq.test.listener.AbstractListener;
public class BaseBroadCastIT extends BaseConf {
private static Logger logger = Logger.getLogger(BaseBroadCastIT.class);
public class BaseBroadcast extends BaseConf {
private static Logger logger = Logger.getLogger(BaseBroadcast.class);
public static RMQBroadCastConsumer getBroadCastConsumer(String nsAddr, String topic,
String subExpression,
AbstractListener listner) {
AbstractListener listener) {
String consumerGroup = initConsumerGroup();
return getBroadCastConsumer(nsAddr, consumerGroup, topic, subExpression, listner);
return getBroadCastConsumer(nsAddr, consumerGroup, topic, subExpression, listener);
}
public static RMQBroadCastConsumer getBroadCastConsumer(String nsAddr, String consumerGroup,
String topic, String subExpression,
AbstractListener listner) {
AbstractListener listener) {
RMQBroadCastConsumer consumer = ConsumerFactory.getRMQBroadCastConsumer(nsAddr,
consumerGroup, topic, subExpression, listner);
consumerGroup, topic, subExpression, listener);
consumer.setDebug();
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.normal;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -30,7 +30,7 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastNormalMsgNotRecvIT extends BaseBroadCastIT {
public class BroadcastNormalMsgNotReceiveIT extends BaseBroadcast {
private static Logger logger = Logger
.getLogger(NormalMsgTwoSameGroupConsumerIT.class);
private RMQNormalProducer producer = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.normal;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,7 +31,7 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastNormalMsgRecvCrashIT extends BaseBroadCastIT {
public class BroadcastNormalMsgRecvCrashIT extends BaseBroadcast {
private static Logger logger = Logger
.getLogger(NormalMsgTwoSameGroupConsumerIT.class);
private RMQNormalProducer producer = null;
@@ -19,7 +19,7 @@ package org.apache.rocketmq.test.client.consumer.broadcast.normal;
import org.apache.log4j.Logger;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -32,7 +32,7 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastNormalMsgRecvFailIT extends BaseBroadCastIT {
public class BroadcastNormalMsgRecvFailIT extends BaseBroadcast {
private static Logger logger = Logger
.getLogger(NormalMsgTwoSameGroupConsumerIT.class);
private RMQNormalProducer producer = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.normal;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,7 +31,7 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastNormalMsgRecvStartLaterIT extends BaseBroadCastIT {
public class BroadcastNormalMsgRecvStartLaterIT extends BaseBroadcast {
private static Logger logger = Logger
.getLogger(NormalMsgTwoSameGroupConsumerIT.class);
private RMQNormalProducer producer = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.normal;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,7 +31,7 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastNormalMsgTwoDiffGroupRecvIT extends BaseBroadCastIT {
public class BroadcastNormalMsgTwoDiffGroupRecvIT extends BaseBroadcast {
private static Logger logger = Logger
.getLogger(NormalMsgTwoSameGroupConsumerIT.class);
private RMQNormalProducer producer = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.normal;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,7 +31,7 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class NormalMsgTwoSameGroupConsumerIT extends BaseBroadCastIT {
public class NormalMsgTwoSameGroupConsumerIT extends BaseBroadcast {
private static Logger logger = Logger
.getLogger(NormalMsgTwoSameGroupConsumerIT.class);
private RMQNormalProducer producer = null;
@@ -20,7 +20,7 @@ package org.apache.rocketmq.test.client.consumer.broadcast.order;
import java.util.List;
import org.apache.log4j.Logger;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.order.RMQOrderListener;
@@ -38,8 +38,8 @@ import static com.google.common.truth.Truth.assertThat;
* Currently, dose not support the ordered broadcast message
*/
@Ignore
public class OrderMsgBroadCastIT extends BaseBroadCastIT {
private static Logger logger = Logger.getLogger(OrderMsgBroadCastIT.class);
public class OrderMsgBroadcastIT extends BaseBroadcast {
private static Logger logger = Logger.getLogger(OrderMsgBroadcastIT.class);
private RMQNormalProducer producer = null;
private String topic = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.tag;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,8 +31,8 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastTwoConsumerFilterIT extends BaseBroadCastIT {
private static Logger logger = Logger.getLogger(BroadCastTwoConsumerSubTagIT.class);
public class BroadcastTwoConsumerFilterIT extends BaseBroadcast {
private static Logger logger = Logger.getLogger(BroadcastTwoConsumerSubTagIT.class);
private RMQNormalProducer producer = null;
private String topic = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.tag;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,8 +31,8 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastTwoConsumerSubDiffTagIT extends BaseBroadCastIT {
private static Logger logger = Logger.getLogger(BroadCastTwoConsumerSubTagIT.class);
public class BroadcastTwoConsumerSubDiffTagIT extends BaseBroadcast {
private static Logger logger = Logger.getLogger(BroadcastTwoConsumerSubTagIT.class);
private RMQNormalProducer producer = null;
private String topic = null;
@@ -18,7 +18,7 @@
package org.apache.rocketmq.test.client.consumer.broadcast.tag;
import org.apache.log4j.Logger;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadcast;
import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
@@ -31,8 +31,8 @@ import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
public class BroadCastTwoConsumerSubTagIT extends BaseBroadCastIT {
private static Logger logger = Logger.getLogger(BroadCastTwoConsumerSubTagIT.class);
public class BroadcastTwoConsumerSubTagIT extends BaseBroadcast {
private static Logger logger = Logger.getLogger(BroadcastTwoConsumerSubTagIT.class);
private RMQNormalProducer producer = null;
private String topic = null;
@@ -140,7 +140,11 @@ public class GrpcBaseIT extends BaseConf {
if (mockProxyHomeURL != null) {
mockProxyHome = mockProxyHomeURL.toURI().getPath();
}
System.setProperty(RMQ_PROXY_HOME, mockProxyHome);
if (null != mockProxyHome) {
System.setProperty(RMQ_PROXY_HOME, mockProxyHome);
}
ConfigurationManager.initEnv();
ConfigurationManager.intConfig();
ConfigurationManager.getProxyConfig().setGrpcServerPort(port);
+68
View File
@@ -0,0 +1,68 @@
#
# 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.
#
load("//bazel:GenTestRules.bzl", "GenTestRules")
java_library(
name = "tools",
srcs = glob(["src/main/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
"//acl",
"//remoting",
"//logging",
"//client",
"//common",
"//srvutil",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:commons_validator_commons_validator",
"@maven//:com_github_luben_zstd_jni",
"@maven//:org_lz4_lz4_java",
"@maven//:com_alibaba_fastjson",
"@maven//:io_netty_netty_all",
"@maven//:commons_cli_commons_cli",
"@maven//:org_slf4j_slf4j_api",
"@maven//:ch_qos_logback_logback_classic",
"@maven//:ch_qos_logback_logback_core",
"@maven//:commons_collections_commons_collections",
],
)
java_library(
name = "tests",
srcs = glob(["src/test/java/**/*.java"]),
visibility = ["//visibility:public"],
deps = [
":tools",
"//client",
"//common",
"//srvutil",
"//remoting",
"//:test_deps",
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:commons_cli_commons_cli",
],
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",
],
)
@@ -18,7 +18,6 @@ package org.apache.rocketmq.tools.command;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
import java.util.ArrayList;
import java.util.List;
import java.nio.file.Files;
@@ -268,7 +267,7 @@ public class MQAdminStartup {
initCommand(new ReElectMasterSubCommand());
}
private static void initLogback() throws JoranException {
private static void initLogback() throws Exception {
LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
JoranConfigurator configurator = new JoranConfigurator();
configurator.setContext(lc);
@@ -80,7 +80,7 @@ public class ClusterAclConfigVersionListSubCommand implements SubCommand {
defaultMQAdminExt.start();
printClusterBaseInfo(defaultMQAdminExt, addr);
System.out.printf("get broker's plain access config version success.%n", addr);
System.out.printf("get broker's plain access config version success. Address:%s %n", addr);
return;
} else if (commandLine.hasOption('c')) {
@@ -58,6 +58,10 @@ public class ProducerConnectionSubCommand implements SubCommand {
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
try {
defaultMQAdminExt.start();
@@ -99,6 +99,10 @@ public class ConsumerProgressSubCommand implements SubCommand {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
try {
defaultMQAdminExt.start();
@@ -77,6 +77,10 @@ public class ConsumerStatusSubCommand implements SubCommand {
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
try {
defaultMQAdminExt.start();
String group = commandLine.getOptionValue('g').trim();
@@ -60,6 +60,11 @@ public class GetConsumerConfigSubCommand implements SubCommand {
DefaultMQAdminExt adminExt = new DefaultMQAdminExt(rpcHook);
adminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
String groupName = commandLine.getOptionValue('g').trim();
if (commandLine.hasOption('n')) {
adminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
try {
adminExt.start();
List<ConsumerConfigInfo> consumerConfigInfoList = new ArrayList<>();
@@ -76,6 +76,9 @@ public class QueryMsgTraceByIdSubCommand implements SubCommand {
if (commandLine.hasOption('t')) {
traceTopic = commandLine.getOptionValue('t').trim();
}
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
this.queryTraceByMsgId(defaultMQAdminExt, traceTopic, msgId);
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + "command failed", e);
@@ -63,6 +63,10 @@ public class UpdateKvConfigCommand implements SubCommand {
// key name
String value = commandLine.getOptionValue('v').trim();
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
defaultMQAdminExt.start();
defaultMQAdminExt.createAndUpdateKvConfig(namespace, key, value);
System.out.printf("create or update kv config to namespace success.%n");
@@ -67,6 +67,11 @@ public class GetConsumerStatusCommand implements SubCommand {
if (commandLine.hasOption("i")) {
originClientId = commandLine.getOptionValue("i").trim();
}
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
defaultMQAdminExt.start();
Map<String, Map<MessageQueue, Long>> consumerStatusTable =
@@ -114,6 +114,10 @@ public class ResetOffsetByTimeCommand implements SubCommand {
queueId = Integer.valueOf(commandLine.getOptionValue('q'));
}
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
defaultMQAdminExt.start();
if (brokerAddr != null && queueId > -1) {
@@ -481,6 +481,6 @@ public class DefaultMQAdminExtTest {
@Test
public void testExamineTopicConfig() throws MQBrokerException, RemotingException, InterruptedException {
TopicConfig topicConfig = defaultMQAdminExt.examineTopicConfig("127.0.0.1:10911", "topic_test_examine_topicConfig");
assertThat(topicConfig.getTopicName().equals("topic_test_examine_topicConfig"));
assertThat(topicConfig.getTopicName().equals("topic_test_examine_topicConfig")).isTrue();
}
}

Some files were not shown because too many files have changed in this diff Show More