opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4873]Optimize controller module test cases execution time (#4882)

Browse Source
This commit is contained in:
mxsm
2022-08-25 14:43:44 +08:00
committed by GitHub
parent 0bd8bdf7f9
commit 5578808b44
2 changed files with 58 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/src/test/java/org/apache/rocketmq/controller/impl/controller/ControllerManagerTest.java
+33 -29
View File
@@ -17,21 +17,23 @@
package org.apache.rocketmq.controller.impl.controller;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.header.namesrv.BrokerHeartbeatRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerResponseHeader;
import org.apache.rocketmq.controller.ControllerManager;
import org.apache.rocketmq.controller.impl.DLedgerController;
import org.apache.rocketmq.remoting.RemotingClient;
@@ -45,7 +47,9 @@ import org.junit.Test;
import static org.apache.rocketmq.common.protocol.ResponseCode.CONTROLLER_NOT_LEADER;
import static org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode.SUCCESS;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class ControllerManagerTest {
private List<String> baseDirs;
@@ -54,7 +58,7 @@ public class ControllerManagerTest {
private NettyRemotingClient remotingClient1;
public ControllerManager launchManager(final String group, final String peers, final String selfId) {
final String path = "/tmp" + File.separator + group + File.separator + selfId;
final String path = System.getProperty("java.io.tmpdir") + File.separator + group + File.separator + selfId;
baseDirs.add(path);
final ControllerConfig config = new ControllerConfig();
@@ -64,7 +68,7 @@ public class ControllerManagerTest {
config.setControllerStorePath(path);
config.setMappedFileSize(10 * 1024 * 1024);
config.setEnableElectUncleanMaster(true);
config.setScanNotActiveBrokerInterval(2000L);
config.setScanNotActiveBrokerInterval(1000L);
final NettyServerConfig serverConfig = new NettyServerConfig();
@@ -90,18 +94,19 @@ public class ControllerManagerTest {
return null;
}
DLedgerController c1 = (DLedgerController) controllers.get(0).getController();
while (c1.getMemberState().getLeaderId() == null) {
Thread.sleep(1000);
}
String leaderId = c1.getMemberState().getLeaderId();
System.out.println("New leader " + leaderId);
for (ControllerManager controllerManager : controllers) {
final DLedgerController controller = (DLedgerController) controllerManager.getController();
if (controller.getMemberState().getSelfId().equals(leaderId)) {
return controllerManager;
ControllerManager manager = await().atMost(Duration.ofSeconds(10)).until(() -> {
String leaderId = c1.getMemberState().getLeaderId();
for (ControllerManager controllerManager : controllers) {
final DLedgerController controller = (DLedgerController) controllerManager.getController();
if (controller.getMemberState().getSelfId().equals(leaderId)) {
System.out.println("New leader " + leaderId);
return controllerManager;
}
}
}
return null;
return null;
}, item -> item != null);
return manager;
}
public void mockData() {
@@ -117,11 +122,11 @@ public class ControllerManagerTest {
*/
public RegisterBrokerToControllerResponseHeader registerBroker(
final String controllerAddress, final String clusterName,
final String brokerName, final String address, final RemotingClient client) throws Exception {
final String brokerName, final String address, final RemotingClient client, final long heartbeatTimeoutMillis) throws Exception {
final RegisterBrokerToControllerRequestHeader requestHeader = new RegisterBrokerToControllerRequestHeader(clusterName, brokerName, address);
// Timeout = 3000
requestHeader.setHeartbeatTimeoutMillis(4000L);
requestHeader.setHeartbeatTimeoutMillis(heartbeatTimeoutMillis);
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_REGISTER_BROKER, requestHeader);
final RemotingCommand response = client.invokeSync(controllerAddress, request, 3000);
assert response != null;
@@ -143,11 +148,11 @@ public class ControllerManagerTest {
String leaderAddr = "localhost" + ":" + leader.getController().getRemotingServer().localListenPort();
// Register two broker, the first one is master.
final RegisterBrokerToControllerResponseHeader responseHeader1 = registerBroker(leaderAddr, "cluster1", "broker1", "127.0.0.1:8000", this.remotingClient);
final RegisterBrokerToControllerResponseHeader responseHeader1 = registerBroker(leaderAddr, "cluster1", "broker1", "127.0.0.1:8000", this.remotingClient,1000L);
assert responseHeader1 != null;
assertEquals(responseHeader1.getBrokerId(), MixAll.MASTER_ID);
final RegisterBrokerToControllerResponseHeader responseHeader2 = registerBroker(leaderAddr, "cluster1", "broker1", "127.0.0.1:8001", this.remotingClient1);
final RegisterBrokerToControllerResponseHeader responseHeader2 = registerBroker(leaderAddr, "cluster1", "broker1", "127.0.0.1:8001", this.remotingClient1,4000L);
assert responseHeader2 != null;
assertEquals(responseHeader2.getBrokerId(), 2);
@@ -166,17 +171,16 @@ public class ControllerManagerTest {
e.printStackTrace();
}
}, 0, 2000L, TimeUnit.MILLISECONDS);
// Wait until the master is expired.
Thread.sleep(6000);
Boolean flag = await().atMost(Duration.ofSeconds(5)).until(() -> {
final GetReplicaInfoRequestHeader requestHeader = new GetReplicaInfoRequestHeader("broker1");
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_REPLICA_INFO, requestHeader);
final RemotingCommand response = this.remotingClient1.invokeSync(leaderAddr, request, 3000);
final GetReplicaInfoResponseHeader responseHeader = (GetReplicaInfoResponseHeader) response.decodeCommandCustomHeader(GetReplicaInfoResponseHeader.class);
return StringUtils.equals(responseHeader.getMasterAddress(), "127.0.0.1:8001");
}, item -> item);
// The new master should be broker2.
final GetReplicaInfoRequestHeader requestHeader = new GetReplicaInfoRequestHeader("broker1");
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_REPLICA_INFO, requestHeader);
final RemotingCommand response = this.remotingClient1.invokeSync(leaderAddr, request, 3000);
final GetReplicaInfoResponseHeader responseHeader = (GetReplicaInfoResponseHeader) response.decodeCommandCustomHeader(GetReplicaInfoResponseHeader.class);
assertEquals(responseHeader.getMasterAddress(), "127.0.0.1:8001");
assertTrue(flag);
executor.shutdown();
}
controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest.java
+25 -16
View File
@@ -18,6 +18,7 @@ package org.apache.rocketmq.controller.impl.controller.impl;
import io.openmessaging.storage.dledger.DLedgerConfig;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -26,14 +27,15 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.SyncStateSet;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerToControllerResponseHeader;
import org.apache.rocketmq.controller.Controller;
import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
import org.apache.rocketmq.controller.impl.DLedgerController;
@@ -43,6 +45,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -124,17 +127,17 @@ public class DLedgerControllerTest {
return null;
}
DLedgerController c1 = controllers.get(0);
while (c1.getMemberState().getLeaderId() == null) {
Thread.sleep(1000);
}
String leaderId = c1.getMemberState().getLeaderId();
System.out.println("New leader " + leaderId);
for (DLedgerController controller : controllers) {
if (controller.getMemberState().getSelfId().equals(leaderId)) {
return controller;
DLedgerController dLedgerController = await().atMost(Duration.ofSeconds(10)).until(() -> {
String leaderId = c1.getMemberState().getLeaderId();
for (DLedgerController controller : controllers) {
if (controller.getMemberState().getSelfId().equals(leaderId)) {
System.out.println("New leader " + leaderId);
return controller;
}
}
}
return null;
return null;
}, item -> item != null);
return dLedgerController;
}
public DLedgerController mockMetaData(boolean enableElectUncleanMaster) throws Exception {
@@ -271,16 +274,22 @@ public class DLedgerControllerTest {
public void testChangeControllerLeader() throws Exception {
final DLedgerController leader = mockMetaData(false);
leader.shutdown();
Thread.sleep(2000);
this.controllers.remove(leader);
// Wait leader again
final DLedgerController newLeader = waitLeader(this.controllers);
assertNotNull(newLeader);
final RemotingCommand resp = newLeader.getReplicaInfo(new GetReplicaInfoRequestHeader("broker1")).get(10, TimeUnit.SECONDS);
final GetReplicaInfoResponseHeader replicaInfo = (GetReplicaInfoResponseHeader) resp.readCustomHeader();
final SyncStateSet syncStateSetResult = RemotingSerializable.decode(resp.getBody(), SyncStateSet.class);
RemotingCommand response = await().atMost(Duration.ofSeconds(10)).until(() -> {
final RemotingCommand resp = newLeader.getReplicaInfo(new GetReplicaInfoRequestHeader("broker1")).get(10, TimeUnit.SECONDS);
if (resp.getCode() == ResponseCode.SUCCESS) {
return resp;
}
return null;
}, item -> item != null);
final GetReplicaInfoResponseHeader replicaInfo = (GetReplicaInfoResponseHeader) response.readCustomHeader();
final SyncStateSet syncStateSetResult = RemotingSerializable.decode(response.getBody(), SyncStateSet.class);
assertEquals(replicaInfo.getMasterAddress(), "127.0.0.1:9000");
assertEquals(replicaInfo.getMasterEpoch(), 1);