fix: server switch not work when only 1 server in cluster (#264)

andrewshan · web-flow · commit fcf6867640a1 · 2023-01-17T19:34:30.000+08:00
diff --git a/polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/pom.xml b/polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/pom.xml
@@ -43,5 +43,17 @@
             <version>${slf4j.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.tencent.polaris</groupId>
+            <artifactId>polaris-test-mock-discovery</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.tencent.polaris</groupId>
+            <artifactId>polaris-test-common</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/src/main/java/com/tencent/polaris/plugins/connector/grpc/ConnectionManager.java b/polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/src/main/java/com/tencent/polaris/plugins/connector/grpc/ConnectionManager.java
@@ -20,6 +20,7 @@
 import com.tencent.polaris.api.config.Configuration;
 import com.tencent.polaris.api.config.global.ClusterType;
 import com.tencent.polaris.api.config.global.ServerConnectorConfig;
+import com.tencent.polaris.api.config.verify.DefaultValues;
 import com.tencent.polaris.api.control.Destroyable;
 import com.tencent.polaris.api.exception.ErrorCode;
 import com.tencent.polaris.api.exception.PolarisException;
@@ -50,6 +51,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import org.slf4j.Logger;
 
 /**
@@ -74,6 +76,7 @@ public class ConnectionManager extends Destroyable {
     private final String clientId;
     private Extensions extensions;
     private final ChannelTlsCertificates tlsCertificates;
+    private Consumer<ConnID> callbackOnSwitched;
 
     /**
      * 构造器
@@ -159,6 +162,15 @@ public boolean checkReady(ClusterType clusterType) {
         return serverAddressList.ready.get();
     }
 
+    public Consumer<ConnID> getCallbackOnSwitched() {
+        return callbackOnSwitched;
+    }
+
+    public void setCallbackOnSwitched(
+            Consumer<ConnID> callbackOnSwitched) {
+        this.callbackOnSwitched = callbackOnSwitched;
+    }
+
     /**
      * 设置准备状态
      *
@@ -257,9 +269,6 @@ private class SwitchServerTask implements Runnable {
         public void run() {
             for (Map.Entry<ClusterType, ServerAddressList> entry : serverAddresses.entrySet()) {
                 ClusterType clusterType = entry.getKey();
-                if (clusterType == ClusterType.BUILTIN_CLUSTER) {
-                    continue;
-                }
                 try {
                     ServerAddressList serverAddressList = entry.getValue();
                     serverAddressList.switchClient();
@@ -399,21 +408,35 @@ public void switchClientOnFail(ConnID lastConn) throws PolarisException {
                     //已经完成切换，不处理
                     return;
                 }
-                Node servAddress = getServerAddress();
-                if (null == servAddress) {
-                    return;
-                }
-                if (null != curConnection) {
-                    if (servAddress.getHost().equals(curConnection.getConnID().getHost())
-                            && servAddress.getPort() == curConnection.getConnID().getPort()) {
-                        return;
-                    }
-                    curConnection.lazyClose();
-                }
-                ConnID connID = new ConnID(serverServiceInfo.getServiceKey(), clusterType, servAddress.getHost(),
-                        servAddress.getPort(), protocol);
-                Connection connection = connectTarget(connID);
-                curConnectionValue.set(connection);
+                doSwitchClient(curConnection);
+            }
+        }
+
+        private void doSwitchClient(Connection curConnection) throws PolarisException {
+            Node servAddress = getServerAddress();
+            if (null == servAddress) {
+                return;
+            }
+            String preAddress = null;
+            if (null != curConnection) {
+                curConnection.lazyClose();
+                preAddress = String.format(
+                        "%s:%d", curConnection.getConnID().getHost(), curConnection.getConnID().getPort());
+            }
+            String namespace = DefaultValues.DEFAULT_SYSTEM_NAMESPACE;
+            String serviceName = DefaultValues.DEFAULT_BUILTIN_DISCOVER;
+            if (null != serverServiceInfo) {
+                namespace = serverServiceInfo.getServiceKey().getNamespace();
+                serviceName = serverServiceInfo.getServiceKey().getService();
+            }
+            ConnID connID = new ConnID(new ServiceKey(namespace, serviceName), clusterType, servAddress.getHost(),
+                    servAddress.getPort(), protocol);
+            Connection connection = connectTarget(connID);
+            curConnectionValue.set(connection);
+            LOG.info("server {} connection switched from {} to {}:{}",
+                    serviceName, preAddress, servAddress.getHost(), servAddress.getPort());
+            if (null != callbackOnSwitched) {
+                callbackOnSwitched.accept(connection.getConnID());
             }
         }
 
@@ -423,25 +446,12 @@ public void switchClient() throws PolarisException {
             if (!Connection.isAvailableConnection(curConnection)) {
                 return;
             }
-            LOG.info("start switch for {}", serverServiceInfo.getServiceKey());
             synchronized (lock) {
                 curConnection = curConnectionValue.get();
                 if (!Connection.isAvailableConnection(curConnection)) {
                     return;
                 }
-                Node servAddress = getServerAddress();
-                if (null == servAddress) {
-                    return;
-                }
-                if (servAddress.getHost().equals(curConnection.getConnID().getHost())
-                        && servAddress.getPort() == curConnection.getConnID().getPort()) {
-                    return;
-                }
-                ConnID connID = new ConnID(serverServiceInfo.getServiceKey(), clusterType, servAddress.getHost(),
-                        servAddress.getPort(), protocol);
-                Connection connection = connectTarget(connID);
-                curConnection.lazyClose();
-                curConnectionValue.set(connection);
+                doSwitchClient(curConnection);
             }
         }
 
diff --git a/polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/src/test/java/com/tencent/polaris/plugins/connector/grpc/ConnectionManagerTest.java b/polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/src/test/java/com/tencent/polaris/plugins/connector/grpc/ConnectionManagerTest.java
@@ -0,0 +1,106 @@
+/*
+ * Tencent is pleased to support the open source community by making Polaris available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * 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.
+ */
+
+package com.tencent.polaris.plugins.connector.grpc;
+
+import static com.tencent.polaris.test.common.Consts.NAMESPACE_TEST;
+import static com.tencent.polaris.test.common.Consts.SERVICE_CIRCUIT_BREAKER;
+import static com.tencent.polaris.test.common.Consts.SERVICE_PROVIDER;
+
+import com.tencent.polaris.api.config.Configuration;
+import com.tencent.polaris.api.config.global.ClusterType;
+import com.tencent.polaris.api.config.verify.DefaultValues;
+import com.tencent.polaris.api.plugin.common.PluginTypes;
+import com.tencent.polaris.api.plugin.compose.Extensions;
+import com.tencent.polaris.api.plugin.server.ServerConnector;
+import com.tencent.polaris.api.pojo.ServiceKey;
+import com.tencent.polaris.client.api.BaseEngine;
+import com.tencent.polaris.client.api.SDKContext;
+import com.tencent.polaris.factory.config.ConfigurationImpl;
+import com.tencent.polaris.plugins.connector.grpc.Connection.ConnID;
+import com.tencent.polaris.test.common.TestUtils;
+import com.tencent.polaris.test.mock.discovery.NamingServer;
+import com.tencent.polaris.test.mock.discovery.NamingService.InstanceParameter;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ConnectionManagerTest {
+
+    private NamingServer namingServer;
+
+    @Before
+    public void before() {
+        try {
+            namingServer = NamingServer.startNamingServer(-1);
+            System.setProperty(TestUtils.SERVER_ADDRESS_ENV, String.format("127.0.0.1:%d", namingServer.getPort()));
+        } catch (IOException e) {
+            Assert.fail(e.getMessage());
+        }
+        ServiceKey serviceKey = new ServiceKey(NAMESPACE_TEST, SERVICE_PROVIDER);
+        InstanceParameter parameter = new InstanceParameter();
+        parameter.setHealthy(true);
+        parameter.setIsolated(false);
+        parameter.setWeight(100);
+        namingServer.getNamingService().batchAddInstances(serviceKey, 10012, 1, parameter);
+    }
+
+    @After
+    public void after() {
+        if (null != namingServer) {
+            namingServer.terminate();
+        }
+    }
+
+    @Test
+    public void testSwitchClient() {
+        Configuration configuration = TestUtils.configWithEnvAddress();
+        ((ConfigurationImpl)configuration).getGlobal().getServerConnector().setServerSwitchInterval(1000L);
+        AtomicBoolean switched = new AtomicBoolean(false);
+        try (SDKContext sdkContext = SDKContext.initContextByConfig(configuration)) {
+            ServerConnector serverConnector = (ServerConnector) sdkContext.getPlugins().getPlugin(
+                    PluginTypes.SERVER_CONNECTOR.getBaseType(), DefaultValues.DEFAULT_DISCOVER_PROTOCOL);
+            GrpcConnector grpcConnector = (GrpcConnector) serverConnector;
+            ConnectionManager connectionManager = grpcConnector.getConnectionManager();
+            Extensions extensions = new Extensions();
+            connectionManager.setExtensions(extensions);
+            connectionManager.setCallbackOnSwitched(new Consumer<ConnID>() {
+                @Override
+                public void accept(ConnID connID) {
+                    if (switched.compareAndSet(false, true)) {
+                        System.out.println("server switched to " + connID);
+                    }
+                }
+            });
+            Connection testConn = connectionManager.getConnection("test", ClusterType.BUILTIN_CLUSTER);
+            Assert.assertNotNull(testConn);
+            try {
+                Thread.sleep(5000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        } catch (Throwable e) {
+            e.printStackTrace();
+        }
+        Assert.assertTrue(switched.get());
+    }
+
+}
diff --git a/pom.xml b/pom.xml
@@ -64,7 +64,7 @@
 
     <properties>
         <!-- Project revision -->
-        <revision>1.9.2</revision>
+        <revision>1.9.3</revision>
 
         <timestamp>${maven.build.timestamp}</timestamp>
         <skip.maven.deploy>false</skip.maven.deploy>
