opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4889]Fix controller test:DLedgerControllerTest#testChangeControllerLeader (#4892)

Browse Source
This commit is contained in:
mxsm
2022-08-26 09:06:02 +08:00
committed by GitHub
parent 5578808b44
commit 64c18fb0db
2 changed files with 36 additions and 11 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
+10 -5
View File
@@ -58,7 +58,8 @@ public class ControllerManagerTest {
private NettyRemotingClient remotingClient1;
public ControllerManager launchManager(final String group, final String peers, final String selfId) {
final String path = System.getProperty("java.io.tmpdir") + File.separator + group + File.separator + selfId;
String tmpdir = System.getProperty("java.io.tmpdir");
final String path = (StringUtils.endsWith(tmpdir, File.separator) ? tmpdir : tmpdir + File.separator) + group + File.separator + selfId;
baseDirs.add(path);
final ControllerConfig config = new ControllerConfig();
@@ -97,9 +98,12 @@ public class ControllerManagerTest {
ControllerManager manager = await().atMost(Duration.ofSeconds(10)).until(() -> {
String leaderId = c1.getMemberState().getLeaderId();
if (null == leaderId) {
return null;
}
for (ControllerManager controllerManager : controllers) {
final DLedgerController controller = (DLedgerController) controllerManager.getController();
if (controller.getMemberState().getSelfId().equals(leaderId)) {
if (controller.getMemberState().getSelfId().equals(leaderId) && controller.isLeaderState()) {
System.out.println("New leader " + leaderId);
return controllerManager;
}
@@ -122,7 +126,8 @@ public class ControllerManagerTest {
*/
public RegisterBrokerToControllerResponseHeader registerBroker(
final String controllerAddress, final String clusterName,
final String brokerName, final String address, final RemotingClient client, final long heartbeatTimeoutMillis) 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
@@ -148,11 +153,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,1000L);
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,4000L);
final RegisterBrokerToControllerResponseHeader responseHeader2 = registerBroker(leaderAddr, "cluster1", "broker1", "127.0.0.1:8001", this.remotingClient1, 4000L);
assert responseHeader2 != null;
assertEquals(responseHeader2.getBrokerId(), 2);
controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest.java
+26 -6
View File
@@ -25,7 +25,10 @@ import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.SyncStateSet;
@@ -59,7 +62,8 @@ public class DLedgerControllerTest {
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;
String tmpdir = System.getProperty("java.io.tmpdir");
final String path = (StringUtils.endsWith(tmpdir, File.separator) ? tmpdir : tmpdir + File.separator) + group + File.separator + selfId;
baseDirs.add(path);
final ControllerConfig config = new ControllerConfig();
@@ -96,9 +100,20 @@ public class DLedgerControllerTest {
public boolean registerNewBroker(Controller leader, String clusterName, String brokerName, String brokerAddress,
boolean isFirstRegisteredBroker) throws Exception {
// Register new broker
final RegisterBrokerToControllerRequestHeader registerRequest =
new RegisterBrokerToControllerRequestHeader(clusterName, brokerName, brokerAddress);
final RemotingCommand response = leader.registerBroker(registerRequest).get(10, TimeUnit.SECONDS);
final RegisterBrokerToControllerRequestHeader registerRequest = new RegisterBrokerToControllerRequestHeader(clusterName, brokerName, brokerAddress);
RemotingCommand response = await().atMost(Duration.ofSeconds(20)).until(() -> {
try {
final RemotingCommand responseInner = leader.registerBroker(registerRequest).get(2, TimeUnit.SECONDS);
if (responseInner == null || responseInner.getCode() != ResponseCode.SUCCESS) {
return null;
}
return responseInner;
} catch (Exception e) {
e.printStackTrace();
return null;
}
}, item -> item != null);
final RegisterBrokerToControllerResponseHeader registerResult = (RegisterBrokerToControllerResponseHeader) response.readCustomHeader();
System.out.println("------------- Register broker done, the result is :" + registerResult);
@@ -113,7 +128,9 @@ public class DLedgerControllerTest {
final AlterSyncStateSetRequestHeader alterRequest =
new AlterSyncStateSetRequestHeader(brokerName, masterAddress, masterEpoch);
final RemotingCommand response = leader.alterSyncStateSet(alterRequest, new SyncStateSet(newSyncStateSet, syncStateSetEpoch)).get(10, TimeUnit.SECONDS);
if (null == response || response.getCode() != ResponseCode.SUCCESS) {
return false;
}
final RemotingCommand getInfoResponse = leader.getReplicaInfo(new GetReplicaInfoRequestHeader(brokerName)).get(10, TimeUnit.SECONDS);
final GetReplicaInfoResponseHeader replicaInfo = (GetReplicaInfoResponseHeader) getInfoResponse.readCustomHeader();
final SyncStateSet syncStateSet = RemotingSerializable.decode(getInfoResponse.getBody(), SyncStateSet.class);
@@ -129,8 +146,11 @@ public class DLedgerControllerTest {
DLedgerController c1 = controllers.get(0);
DLedgerController dLedgerController = await().atMost(Duration.ofSeconds(10)).until(() -> {
String leaderId = c1.getMemberState().getLeaderId();
if (null == leaderId) {
return null;
}
for (DLedgerController controller : controllers) {
if (controller.getMemberState().getSelfId().equals(leaderId)) {
if (controller.getMemberState().getSelfId().equals(leaderId) && controller.isLeaderState()) {
System.out.println("New leader " + leaderId);
return controller;
}