From 64c18fb0dbb8d7e0e4af9edf49d04ab8dd64bccd Mon Sep 17 00:00:00 2001 From: mxsm Date: Fri, 26 Aug 2022 09:06:02 +0800 Subject: [PATCH] [ISSUE #4889]Fix controller test:DLedgerControllerTest#testChangeControllerLeader (#4892) --- .../controller/ControllerManagerTest.java | 15 ++++++--- .../impl/DLedgerControllerTest.java | 32 +++++++++++++++---- 2 files changed, 36 insertions(+), 11 deletions(-) diff --git a/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/ControllerManagerTest.java b/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/ControllerManagerTest.java index 46dd0661b1..8da5347d01 100644 --- a/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/ControllerManagerTest.java +++ b/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/ControllerManagerTest.java @@ -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); diff --git a/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest.java b/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest.java index ce9ecf1e8c..00809aacc2 100644 --- a/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest.java +++ b/controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DLedgerControllerTest.java @@ -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; }