[ISSUE #7330] Add goaway and reconnection mechanism

controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java

@@ -63,7 +63,8 @@ public class DLedgerControllerTest {
     private List<String> baseDirs;
     private List<DLedgerController> controllers;
 
-    public DLedgerController launchController(final String group, final String peers, final String selfId, final boolean isEnableElectUncleanMaster) {
+    public DLedgerController launchController(final String group, final String peers, final String selfId,
+        final boolean isEnableElectUncleanMaster) {
         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);
@@ -121,11 +122,11 @@ public void registerNewBroker(Controller leader, String clusterName, String brok
         final RegisterBrokerToControllerRequestHeader registerBrokerToControllerRequestHeader = new RegisterBrokerToControllerRequestHeader(clusterName, brokerName, nextBrokerId, brokerAddress);
         RemotingCommand remotingCommand2 = leader.registerBroker(registerBrokerToControllerRequestHeader).get(2, TimeUnit.SECONDS);
 
-
         assertEquals(ResponseCode.SUCCESS, remotingCommand2.getCode());
     }
 
-    public void brokerTryElectMaster(Controller leader, String clusterName, String brokerName, String brokerAddress, Long brokerId,
+    public void brokerTryElectMaster(Controller leader, String clusterName, String brokerName, String brokerAddress,
+        Long brokerId,
         boolean exceptSuccess) throws Exception {
         final ElectMasterRequestHeader electMasterRequestHeader = ElectMasterRequestHeader.ofBrokerTrigger(clusterName, brokerName, brokerId);
         RemotingCommand command = leader.electMaster(electMasterRequestHeader).get(2, TimeUnit.SECONDS);
@@ -186,9 +187,9 @@ public DLedgerController mockMetaData(boolean enableElectUncleanMaster) throws E
         registerNewBroker(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[1], 2L);
         registerNewBroker(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[2], 3L);
         // try elect
-        brokerTryElectMaster(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[0], 1L,true);
-        brokerTryElectMaster(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[1], 2L,  false);
-        brokerTryElectMaster(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[2], 3L,false);
+        brokerTryElectMaster(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[0], 1L, true);
+        brokerTryElectMaster(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[1], 2L, false);
+        brokerTryElectMaster(leader, DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, DEFAULT_IP[2], 3L, false);
         final RemotingCommand getInfoResponse = leader.getReplicaInfo(new GetReplicaInfoRequestHeader(DEFAULT_BROKER_NAME)).get(10, TimeUnit.SECONDS);
         final GetReplicaInfoResponseHeader replicaInfo = (GetReplicaInfoResponseHeader) getInfoResponse.readCustomHeader();
         assertEquals(1, replicaInfo.getMasterEpoch().intValue());
@@ -239,6 +240,8 @@ public void testElectMaster() throws Exception {
     @Test
     public void testBrokerLifecycleListener() throws Exception {
         final DLedgerController leader = mockMetaData(false);
+
+        assertTrue(leader.isLeaderState());
         // Mock that master broker has been inactive, and try to elect a new master from sync-state-set
         // But we shut down two controller, so the ElectMasterEvent will be appended to DLedger failed.
         // So the statemachine still keep the stale master's information
@@ -247,15 +250,20 @@ public void testBrokerLifecycleListener() throws Exception {
             dLedgerController.shutdown();
             controllers.remove(dLedgerController);
         }
+
         final ElectMasterRequestHeader request = ElectMasterRequestHeader.ofControllerTrigger(DEFAULT_BROKER_NAME);
         setBrokerElectPolicy(leader, 1L);
         Exception exception = null;
+        RemotingCommand remotingCommand = null;
         try {
-            leader.electMaster(request).get(5, TimeUnit.SECONDS);
+            remotingCommand = leader.electMaster(request).get(5, TimeUnit.SECONDS);
         } catch (Exception e) {
             exception = e;
         }
-        assertNotNull(exception);
+
+        assertTrue(exception != null ||
+            remotingCommand != null && remotingCommand.getCode() == ResponseCode.CONTROLLER_NOT_LEADER);
+
         // Shut down leader controller
         leader.shutdown();
         controllers.remove(leader);
@@ -272,7 +280,7 @@ public void testBrokerLifecycleListener() throws Exception {
         setBrokerAlivePredicate(newLeader, 1L);
         // Check if the statemachine is stale
         final RemotingCommand resp = newLeader.getReplicaInfo(new GetReplicaInfoRequestHeader(DEFAULT_BROKER_NAME)).
-                get(10, TimeUnit.SECONDS);
+            get(10, TimeUnit.SECONDS);
         final GetReplicaInfoResponseHeader replicaInfo = (GetReplicaInfoResponseHeader) resp.readCustomHeader();
         assertEquals(1, replicaInfo.getMasterBrokerId().longValue());
         assertEquals(1, replicaInfo.getMasterEpoch().intValue());

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyClientConfig.java

@@ -53,6 +53,12 @@ public class NettyClientConfig {
     private boolean disableCallbackExecutor = false;
     private boolean disableNettyWorkerGroup = false;
 
+    private long maxReconnectIntervalTimeSeconds = 60;
+
+    private boolean enableReconnectForGoAway = true;
+
+    private boolean enableTransparentRetry = true;
+
     public boolean isClientCloseSocketIfTimeout() {
         return clientCloseSocketIfTimeout;
     }
@@ -181,6 +187,30 @@ public void setDisableNettyWorkerGroup(boolean disableNettyWorkerGroup) {
         this.disableNettyWorkerGroup = disableNettyWorkerGroup;
     }
 
+    public long getMaxReconnectIntervalTimeSeconds() {
+        return maxReconnectIntervalTimeSeconds;
+    }
+
+    public void setMaxReconnectIntervalTimeSeconds(long maxReconnectIntervalTimeSeconds) {
+        this.maxReconnectIntervalTimeSeconds = maxReconnectIntervalTimeSeconds;
+    }
+
+    public boolean isEnableReconnectForGoAway() {
+        return enableReconnectForGoAway;
+    }
+
+    public void setEnableReconnectForGoAway(boolean enableReconnectForGoAway) {
+        this.enableReconnectForGoAway = enableReconnectForGoAway;
+    }
+
+    public boolean isEnableTransparentRetry() {
+        return enableTransparentRetry;
+    }
+
+    public void setEnableTransparentRetry(boolean enableTransparentRetry) {
+        this.enableTransparentRetry = enableTransparentRetry;
+    }
+
     public String getSocksProxyConfig() {
         return socksProxyConfig;
     }

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java

@@ -40,9 +40,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import javax.annotation.Nullable;
 import org.apache.rocketmq.common.AbortProcessException;
+import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.UtilAll;
@@ -60,6 +62,7 @@
 import org.apache.rocketmq.remoting.metrics.RemotingMetricsManager;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
 
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_IS_LONG_POLLING;
 import static org.apache.rocketmq.remoting.metrics.RemotingMetricsConstant.LABEL_REQUEST_CODE;
@@ -120,6 +123,8 @@ public abstract class NettyRemotingAbstract {
      */
     protected List<RPCHook> rpcHooks = new ArrayList<>();
 
+    protected AtomicBoolean isShuttingDown = new AtomicBoolean(false);
+
     static {
         NettyLogger.initNettyLogger();
     }
@@ -264,6 +269,16 @@ public void processRequestCommand(final ChannelHandlerContext ctx, final Remotin
 
         Runnable run = buildProcessRequestHandler(ctx, cmd, pair, opaque);
 
+        if (isShuttingDown.get()) {
+            if (cmd.getVersion() > MQVersion.Version.V5_1_4.ordinal()) {
+                final RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.GO_AWAY,
+                    "please go away");
+                response.setOpaque(opaque);
+                writeResponse(ctx.channel(), cmd, response);
+                return;
+            }
+        }
+
         if (pair.getObject1().rejectRequest()) {
             final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                 "[REJECTREQUEST]system busy, start flow control for a while");