Compatible for curator5. (#8263)

* compatible for curator5.

* compatible for curator5.

* fix ut.

* dependency fix.

* for compatible, dubbo-sample CompatibleRegistry use setMethod.

* ut fix.

* use curator as the default ZookeeperTransporter.

* ut fix.

* code clean. make ZookeeperTransporter inject at constructor.

dubbo-compatible/pom.xml

@@ -43,6 +43,11 @@
             <artifactId>dubbo-remoting-zookeeper</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-rpc-rest</artifactId>

dubbo-configcenter/dubbo-configcenter-zookeeper/pom.xml

@@ -38,6 +38,11 @@
             <artifactId>dubbo-remoting-zookeeper</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.curator</groupId>
             <artifactId>curator-test</artifactId>

dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java

@@ -19,6 +19,7 @@ package org.apache.dubbo.configcenter.support.zookeeper;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.config.configcenter.AbstractDynamicConfigurationFactory;
 import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
+import org.apache.dubbo.common.extension.DisableInject;
 import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
 
 /**
@@ -28,6 +29,11 @@ public class ZookeeperDynamicConfigurationFactory extends AbstractDynamicConfigu
 
     private ZookeeperTransporter zookeeperTransporter;
 
+    public ZookeeperDynamicConfigurationFactory() {
+        this.zookeeperTransporter = ZookeeperTransporter.getExtension();
+    }
+
+    @DisableInject
     public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
         this.zookeeperTransporter = zookeeperTransporter;
     }

dubbo-distribution/dubbo-all/pom.xml

@@ -87,6 +87,13 @@
             <scope>compile</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-rpc-api</artifactId>
@@ -404,6 +411,7 @@
                                     <include>org.apache.dubbo:dubbo-remoting-netty</include>
                                     <include>org.apache.dubbo:dubbo-remoting-netty4</include>
                                     <include>org.apache.dubbo:dubbo-remoting-zookeeper</include>
+                                    <include>org.apache.dubbo:dubbo-remoting-zookeeper-curator5</include>
                                     <include>org.apache.dubbo:dubbo-rpc-api</include>
                                     <include>org.apache.dubbo:dubbo-rpc-dubbo</include>
                                     <include>org.apache.dubbo:dubbo-rpc-injvm</include>

dubbo-distribution/dubbo-bom/pom.xml

@@ -93,6 +93,11 @@
                 <artifactId>dubbo-remoting-zookeeper</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.dubbo</groupId>
+                <artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.dubbo</groupId>
                 <artifactId>dubbo-rpc-api</artifactId>

dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReportFactory.java

@@ -17,6 +17,7 @@
 package org.apache.dubbo.metadata.store.zookeeper;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.DisableInject;
 import org.apache.dubbo.metadata.report.MetadataReport;
 import org.apache.dubbo.metadata.report.support.AbstractMetadataReportFactory;
 import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
@@ -28,6 +29,11 @@ public class ZookeeperMetadataReportFactory extends AbstractMetadataReportFactor
 
     private ZookeeperTransporter zookeeperTransporter;
 
+    public ZookeeperMetadataReportFactory() {
+        this.zookeeperTransporter = ZookeeperTransporter.getExtension();
+    }
+
+    @DisableInject
     public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
         this.zookeeperTransporter = zookeeperTransporter;
     }

dubbo-metadata/dubbo-metadata-report-zookeeper/src/test/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReportTest.java

@@ -17,7 +17,6 @@
 package org.apache.dubbo.metadata.store.zookeeper;
 
 import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.extension.ExtensionLoader;
 import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.metadata.definition.ServiceDefinitionBuilder;
 import org.apache.dubbo.metadata.definition.model.FullServiceDefinition;
@@ -26,7 +25,6 @@ import org.apache.dubbo.metadata.report.identifier.KeyTypeEnum;
 import org.apache.dubbo.metadata.report.identifier.MetadataIdentifier;
 import org.apache.dubbo.metadata.report.identifier.ServiceMetadataIdentifier;
 import org.apache.dubbo.metadata.report.identifier.SubscriberMetadataIdentifier;
-import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
 
 import com.google.gson.Gson;
 import org.apache.curator.test.TestingServer;
@@ -60,8 +58,6 @@ public class ZookeeperMetadataReportTest {
         this.registryUrl = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort);
 
         zookeeperMetadataReportFactory = new ZookeeperMetadataReportFactory();
-        ZookeeperTransporter zookeeperTransporter = ExtensionLoader.getExtensionLoader(ZookeeperTransporter.class).getExtension("curator");
-        zookeeperMetadataReportFactory.setZookeeperTransporter(zookeeperTransporter);
         this.zookeeperMetadataReport = (ZookeeperMetadataReport) zookeeperMetadataReportFactory.getMetadataReport(registryUrl);
     }
 

dubbo-registry/dubbo-registry-zookeeper/pom.xml

@@ -40,6 +40,11 @@
             <artifactId>dubbo-remoting-zookeeper</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.curator</groupId>
             <artifactId>curator-x-discovery</artifactId>
@@ -50,4 +55,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-</project>
+</project>

dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistryFactory.java

@@ -17,22 +17,23 @@
 package org.apache.dubbo.registry.zookeeper;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.DisableInject;
 import org.apache.dubbo.registry.Registry;
 import org.apache.dubbo.registry.support.AbstractRegistryFactory;
 import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
 
 /**
  * ZookeeperRegistryFactory.
- *
  */
 public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
 
     private ZookeeperTransporter zookeeperTransporter;
 
-    /**
+    public ZookeeperRegistryFactory() {
-     * Invisible injection of zookeeper client via IOC/SPI
+        this.zookeeperTransporter = ZookeeperTransporter.getExtension();
-     * @param zookeeperTransporter
+    }
-     */
+
+    @DisableInject
     public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
         this.zookeeperTransporter = zookeeperTransporter;
     }

dubbo-registry/dubbo-registry-zookeeper/src/test/java/org/apache/dubbo/registry/zookeeper/ZookeeperRegistryTest.java

@@ -22,7 +22,6 @@ import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.registry.NotifyListener;
 import org.apache.dubbo.registry.Registry;
 import org.apache.dubbo.registry.status.RegistryStatusChecker;
-import org.apache.dubbo.remoting.zookeeper.curator.CuratorZookeeperTransporter;
 
 import org.apache.curator.test.TestingServer;
 import org.junit.jupiter.api.AfterEach;
@@ -59,7 +58,6 @@ public class ZookeeperRegistryTest {
 
         this.registryUrl = URL.valueOf("zookeeper://localhost:" + zkServerPort);
         zookeeperRegistryFactory = new ZookeeperRegistryFactory();
-        zookeeperRegistryFactory.setZookeeperTransporter(new CuratorZookeeperTransporter());
         this.zookeeperRegistry = (ZookeeperRegistry) zookeeperRegistryFactory.createRegistry(registryUrl);
     }
 

dubbo-remoting/dubbo-remoting-api/pom.xml

@@ -51,4 +51,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-</project>
+</project>

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperClient.java

@@ -14,17 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.zookeeper.support;
+package org.apache.dubbo.remoting.zookeeper;
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.config.configcenter.ConfigItem;
 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.common.utils.ConcurrentHashSet;
-import org.apache.dubbo.remoting.zookeeper.ChildListener;
-import org.apache.dubbo.remoting.zookeeper.DataListener;
-import org.apache.dubbo.remoting.zookeeper.StateListener;
-import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
 
 import java.util.List;
 import java.util.Set;

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/AbstractZookeeperTransporter.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.zookeeper.support;
+package org.apache.dubbo.remoting.zookeeper;
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.constants.RemotingConstants;
@@ -22,8 +22,6 @@ import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.common.url.component.ServiceConfigURL;
 import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
-import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -91,7 +89,7 @@ public abstract class AbstractZookeeperTransporter implements ZookeeperTransport
      * @param addressList
      * @return
      */
-    ZookeeperClient fetchAndUpdateZookeeperClientCache(List<String> addressList) {
+    public ZookeeperClient fetchAndUpdateZookeeperClientCache(List<String> addressList) {
 
         ZookeeperClient zookeeperClient = null;
         for (String address : addressList) {
@@ -111,7 +109,7 @@ public abstract class AbstractZookeeperTransporter implements ZookeeperTransport
      * @param url such as:zookeeper://127.0.0.1:2181?127.0.0.1:8989,127.0.0.1:9999
      * @return such as 127.0.0.1:2181,127.0.0.1:8989,127.0.0.1:9999
      */
-    List<String> getURLBackupAddress(URL url) {
+    public List<String> getURLBackupAddress(URL url) {
         List<String> addressList = new ArrayList<String>();
         addressList.add(url.getAddress());
         addressList.addAll(url.getParameter(RemotingConstants.BACKUP_KEY, Collections.EMPTY_LIST));
@@ -177,7 +175,7 @@ public abstract class AbstractZookeeperTransporter implements ZookeeperTransport
      *
      * @return
      */
-    Map<String, ZookeeperClient> getZookeeperClientMap() {
+    public Map<String, ZookeeperClient> getZookeeperClientMap() {
         return zookeeperClientMap;
     }
 }

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/EventType.java

@@ -16,8 +16,6 @@
  */
 package org.apache.dubbo.remoting.zookeeper;
 
-import org.apache.zookeeper.Watcher;
-
 /**
  * 2019-02-26
  */
@@ -33,7 +31,6 @@ public enum EventType {
     INITIALIZED(10);
 
 
-
     private final int intValue;     // Integer representation of value
     // for sending over wire
 
@@ -44,22 +41,4 @@ public enum EventType {
     public int getIntValue() {
         return intValue;
     }
-
-    public static Watcher.Event.EventType fromInt(int intValue) {
-        switch (intValue) {
-            case -1:
-                return Watcher.Event.EventType.None;
-            case 1:
-                return Watcher.Event.EventType.NodeCreated;
-            case 2:
-                return Watcher.Event.EventType.NodeDeleted;
-            case 3:
-                return Watcher.Event.EventType.NodeDataChanged;
-            case 4:
-                return Watcher.Event.EventType.NodeChildrenChanged;
-
-            default:
-                throw new RuntimeException("Invalid integer value for conversion to EventType");
-        }
-    }
 }

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/zookeeper/ZookeeperTransporter.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.dubbo.remoting.zookeeper;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.extension.SPI;
+
+import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoader;
+
+@SPI
+public interface ZookeeperTransporter {
+
+    String CURATOR_5 = "curator5";
+
+    String CURATOR = "curator";
+
+    ZookeeperClient connect(URL url);
+
+    static ZookeeperTransporter getExtension() {
+        ExtensionLoader<ZookeeperTransporter> extensionLoader = getExtensionLoader(ZookeeperTransporter.class);
+        boolean isHighVersion = isHighVersionCurator();
+        if (isHighVersion) {
+            return extensionLoader.getExtension(CURATOR_5);
+        }
+        return extensionLoader.getExtension(CURATOR);
+    }
+
+    static boolean isHighVersionCurator() {
+        try {
+            Class.forName("org.apache.curator.framework.recipes.cache.CuratorCache");
+            return true;
+        } catch (ClassNotFoundException e) {
+            return false;
+        }
+    }
+
+}

dubbo-remoting/dubbo-remoting-zookeeper-curator5/pom.xml

@@ -0,0 +1,69 @@
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dubbo</groupId>
+        <artifactId>dubbo-remoting</artifactId>
+        <version>${revision}</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+    <description>The zookeeper curator5 remoting module of dubbo project</description>
+    <properties>
+        <skip_maven_deploy>false</skip_maven_deploy>
+        <curator5_version>5.0.0</curator5_version>
+        <zookeeper_version>3.6.0</zookeeper_version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-common</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-framework</artifactId>
+            <version>${curator5_version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-recipes</artifactId>
+            <version>${curator5_version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>${zookeeper_version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${curator5_version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/main/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperClient.java

@@ -0,0 +1,438 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.dubbo.remoting.zookeeper.curator5;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.configcenter.ConfigItem;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.remoting.zookeeper.ChildListener;
+import org.apache.dubbo.remoting.zookeeper.DataListener;
+import org.apache.dubbo.remoting.zookeeper.EventType;
+import org.apache.dubbo.remoting.zookeeper.StateListener;
+import org.apache.dubbo.remoting.zookeeper.AbstractZookeeperClient;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.CuratorWatcher;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.NodeCache;
+import org.apache.curator.framework.recipes.cache.NodeCacheListener;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.KeeperException.NodeExistsException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
+
+
+public class Curator5ZookeeperClient extends AbstractZookeeperClient<Curator5ZookeeperClient.NodeCacheListenerImpl, Curator5ZookeeperClient.CuratorWatcherImpl> {
+
+    protected static final Logger logger = LoggerFactory.getLogger(Curator5ZookeeperClient.class);
+    private static final String ZK_SESSION_EXPIRE_KEY = "zk.session.expire";
+
+    private static final Charset CHARSET = StandardCharsets.UTF_8;
+    private final CuratorFramework client;
+    private static Map<String, NodeCache> nodeCacheMap = new ConcurrentHashMap<>();
+
+    public Curator5ZookeeperClient(URL url) {
+        super(url);
+        try {
+            int timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_CONNECTION_TIMEOUT_MS);
+            int sessionExpireMs = url.getParameter(ZK_SESSION_EXPIRE_KEY, DEFAULT_SESSION_TIMEOUT_MS);
+            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+                    .connectString(url.getBackupAddress())
+                    .retryPolicy(new RetryNTimes(1, 1000))
+                    .connectionTimeoutMs(timeout)
+                    .sessionTimeoutMs(sessionExpireMs);
+            String authority = url.getAuthority();
+            if (authority != null && authority.length() > 0) {
+                builder = builder.authorization("digest", authority.getBytes());
+            }
+            client = builder.build();
+            client.getConnectionStateListenable().addListener(new CuratorConnectionStateListener(url));
+            client.start();
+            boolean connected = client.blockUntilConnected(timeout, TimeUnit.MILLISECONDS);
+            if (!connected) {
+                throw new IllegalStateException("zookeeper not connected");
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void createPersistent(String path) {
+        try {
+            client.create().forPath(path);
+        } catch (NodeExistsException e) {
+            logger.warn("ZNode " + path + " already exists.", e);
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void createEphemeral(String path) {
+        try {
+            client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
+        } catch (NodeExistsException e) {
+            logger.warn("ZNode " + path + " already exists, since we will only try to recreate a node on a session expiration" +
+                    ", this duplication might be caused by a delete delay from the zk server, which means the old expired session" +
+                    " may still holds this ZNode and the server just hasn't got time to do the deletion. In this case, " +
+                    "we can just try to delete and create again.", e);
+            deletePath(path);
+            createEphemeral(path);
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void createPersistent(String path, String data) {
+        byte[] dataBytes = data.getBytes(CHARSET);
+        try {
+            client.create().forPath(path, dataBytes);
+        } catch (NodeExistsException e) {
+            try {
+                client.setData().forPath(path, dataBytes);
+            } catch (Exception e1) {
+                throw new IllegalStateException(e.getMessage(), e1);
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void createEphemeral(String path, String data) {
+        byte[] dataBytes = data.getBytes(CHARSET);
+        try {
+            client.create().withMode(CreateMode.EPHEMERAL).forPath(path, dataBytes);
+        } catch (NodeExistsException e) {
+            logger.warn("ZNode " + path + " already exists, since we will only try to recreate a node on a session expiration" +
+                    ", this duplication might be caused by a delete delay from the zk server, which means the old expired session" +
+                    " may still holds this ZNode and the server just hasn't got time to do the deletion. In this case, " +
+                    "we can just try to delete and create again.", e);
+            deletePath(path);
+            createEphemeral(path, data);
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void update(String path, String data, int version) {
+        byte[] dataBytes = data.getBytes(CHARSET);
+        try {
+            client.setData().withVersion(version).forPath(path, dataBytes);
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void createOrUpdatePersistent(String path, String data, int version) {
+        try {
+            if (checkExists(path)) {
+                update(path, data, version);
+            } else {
+                createPersistent(path, data);
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void createOrUpdateEphemeral(String path, String data, int version) {
+        try {
+            if (checkExists(path)) {
+                update(path, data, version);
+            } else {
+                createEphemeral(path, data);
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected void deletePath(String path) {
+        try {
+            client.delete().deletingChildrenIfNeeded().forPath(path);
+        } catch (NoNodeException ignored) {
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public List<String> getChildren(String path) {
+        try {
+            return client.getChildren().forPath(path);
+        } catch (NoNodeException e) {
+            return null;
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public boolean checkExists(String path) {
+        try {
+            if (client.checkExists().forPath(path) != null) {
+                return true;
+            }
+        } catch (Exception ignored) {
+        }
+        return false;
+    }
+
+    @Override
+    public boolean isConnected() {
+        return client.getZookeeperClient().isConnected();
+    }
+
+    @Override
+    public String doGetContent(String path) {
+        try {
+            byte[] dataBytes = client.getData().forPath(path);
+            return (dataBytes == null || dataBytes.length == 0) ? null : new String(dataBytes, CHARSET);
+        } catch (NoNodeException e) {
+            // ignore NoNode Exception.
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+        return null;
+    }
+
+    @Override
+    public ConfigItem doGetConfigItem(String path) {
+        String content;
+        Stat stat;
+        try {
+            stat = new Stat();
+            byte[] dataBytes = client.getData().storingStatIn(stat).forPath(path);
+            content = (dataBytes == null || dataBytes.length == 0) ? null : new String(dataBytes, CHARSET);
+        } catch (NoNodeException e) {
+            return new ConfigItem();
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+        return new ConfigItem(content, stat);
+    }
+
+    @Override
+    public void doClose() {
+        client.close();
+    }
+
+    @Override
+    public Curator5ZookeeperClient.CuratorWatcherImpl createTargetChildListener(String path, ChildListener listener) {
+        return new Curator5ZookeeperClient.CuratorWatcherImpl(client, listener, path);
+    }
+
+    @Override
+    public List<String> addTargetChildListener(String path, CuratorWatcherImpl listener) {
+        try {
+            return client.getChildren().usingWatcher(listener).forPath(path);
+        } catch (NoNodeException e) {
+            return null;
+        } catch (Exception e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    protected Curator5ZookeeperClient.NodeCacheListenerImpl createTargetDataListener(String path, DataListener listener) {
+        return new NodeCacheListenerImpl(client, listener, path);
+    }
+
+    @Override
+    protected void addTargetDataListener(String path, Curator5ZookeeperClient.NodeCacheListenerImpl nodeCacheListener) {
+        this.addTargetDataListener(path, nodeCacheListener, null);
+    }
+
+    @Override
+    protected void addTargetDataListener(String path, Curator5ZookeeperClient.NodeCacheListenerImpl nodeCacheListener, Executor executor) {
+        try {
+            NodeCache nodeCache = new NodeCache(client, path);
+            if (nodeCacheMap.putIfAbsent(path, nodeCache) != null) {
+                return;
+            }
+            if (executor == null) {
+                nodeCache.getListenable().addListener(nodeCacheListener);
+            } else {
+                nodeCache.getListenable().addListener(nodeCacheListener, executor);
+            }
+
+            nodeCache.start();
+        } catch (Exception e) {
+            throw new IllegalStateException("Add nodeCache listener for path:" + path, e);
+        }
+    }
+
+    @Override
+    protected void removeTargetDataListener(String path, Curator5ZookeeperClient.NodeCacheListenerImpl nodeCacheListener) {
+        NodeCache nodeCache = nodeCacheMap.get(path);
+        if (nodeCache != null) {
+            nodeCache.getListenable().removeListener(nodeCacheListener);
+        }
+        nodeCacheListener.dataListener = null;
+    }
+
+    @Override
+    public void removeTargetChildListener(String path, CuratorWatcherImpl listener) {
+        listener.unwatch();
+    }
+
+    static class NodeCacheListenerImpl implements NodeCacheListener {
+
+        private CuratorFramework client;
+
+        private volatile DataListener dataListener;
+
+        private String path;
+
+        protected NodeCacheListenerImpl() {
+        }
+
+        public NodeCacheListenerImpl(CuratorFramework client, DataListener dataListener, String path) {
+            this.client = client;
+            this.dataListener = dataListener;
+            this.path = path;
+        }
+
+        @Override
+        public void nodeChanged() throws Exception {
+            ChildData childData = nodeCacheMap.get(path).getCurrentData();
+            String content = null;
+            EventType eventType;
+            if (childData == null) {
+                eventType = EventType.NodeDeleted;
+            } else {
+                content = new String(childData.getData(), CHARSET);
+                eventType = EventType.NodeDataChanged;
+            }
+            dataListener.dataChanged(path, content, eventType);
+        }
+    }
+
+    static class CuratorWatcherImpl implements CuratorWatcher {
+
+        private CuratorFramework client;
+        private volatile ChildListener childListener;
+        private String path;
+
+        public CuratorWatcherImpl(CuratorFramework client, ChildListener listener, String path) {
+            this.client = client;
+            this.childListener = listener;
+            this.path = path;
+        }
+
+        protected CuratorWatcherImpl() {
+        }
+
+        public void unwatch() {
+            this.childListener = null;
+        }
+
+        @Override
+        public void process(WatchedEvent event) throws Exception {
+            // if client connect or disconnect to server, zookeeper will queue
+            // watched event(Watcher.Event.EventType.None, .., path = null).
+            if (event.getType() == Watcher.Event.EventType.None) {
+                return;
+            }
+
+            if (childListener != null) {
+                childListener.childChanged(path, client.getChildren().usingWatcher(this).forPath(path));
+            }
+        }
+    }
+
+    private class CuratorConnectionStateListener implements ConnectionStateListener {
+        private final long UNKNOWN_SESSION_ID = -1L;
+
+        private long lastSessionId;
+        private int timeout;
+        private int sessionExpireMs;
+
+        public CuratorConnectionStateListener(URL url) {
+            this.timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_CONNECTION_TIMEOUT_MS);
+            this.sessionExpireMs = url.getParameter(ZK_SESSION_EXPIRE_KEY, DEFAULT_SESSION_TIMEOUT_MS);
+        }
+
+        @Override
+        public void stateChanged(CuratorFramework client, ConnectionState state) {
+            long sessionId = UNKNOWN_SESSION_ID;
+            try {
+                sessionId = client.getZookeeperClient().getZooKeeper().getSessionId();
+            } catch (Exception e) {
+                logger.warn("Curator client state changed, but failed to get the related zk session instance.");
+            }
+
+            if (state == ConnectionState.LOST) {
+                logger.warn("Curator zookeeper session " + Long.toHexString(lastSessionId) + " expired.");
+                Curator5ZookeeperClient.this.stateChanged(StateListener.SESSION_LOST);
+            } else if (state == ConnectionState.SUSPENDED) {
+                logger.warn("Curator zookeeper connection of session " + Long.toHexString(sessionId) + " timed out. " +
+                        "connection timeout value is " + timeout + ", session expire timeout value is " + sessionExpireMs);
+                Curator5ZookeeperClient.this.stateChanged(StateListener.SUSPENDED);
+            } else if (state == ConnectionState.CONNECTED) {
+                lastSessionId = sessionId;
+                logger.info("Curator zookeeper client instance initiated successfully, session id is " + Long.toHexString(sessionId));
+                Curator5ZookeeperClient.this.stateChanged(StateListener.CONNECTED);
+            } else if (state == ConnectionState.RECONNECTED) {
+                if (lastSessionId == sessionId && sessionId != UNKNOWN_SESSION_ID) {
+                    logger.warn("Curator zookeeper connection recovered from connection lose, " +
+                            "reuse the old session " + Long.toHexString(sessionId));
+                    Curator5ZookeeperClient.this.stateChanged(StateListener.RECONNECTED);
+                } else {
+                    logger.warn("New session created after old session lost, " +
+                            "old session " + Long.toHexString(lastSessionId) + ", new session " + Long.toHexString(sessionId));
+                    lastSessionId = sessionId;
+                    Curator5ZookeeperClient.this.stateChanged(StateListener.NEW_SESSION_CREATED);
+                }
+            }
+        }
+
+    }
+
+    /**
+     * just for unit test
+     *
+     * @return
+     */
+    CuratorFramework getClient() {
+        return client;
+    }
+}

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/main/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperTransporter.java

@@ -14,17 +14,17 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.zookeeper;
+package org.apache.dubbo.remoting.zookeeper.curator5;
 
 import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.extension.Adaptive;
+import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
-import org.apache.dubbo.common.extension.SPI;
+import org.apache.dubbo.remoting.zookeeper.AbstractZookeeperTransporter;
-import org.apache.dubbo.remoting.Constants;
 
-@SPI("curator")
+public class Curator5ZookeeperTransporter extends AbstractZookeeperTransporter {
-public interface ZookeeperTransporter {
 
-    @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
+    @Override
-    ZookeeperClient connect(URL url);
+    public ZookeeperClient createZookeeperClient(URL url) {
+        return new Curator5ZookeeperClient(url);
+    }
 
 }

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter

@@ -0,0 +1 @@
+curator5=org.apache.dubbo.remoting.zookeeper.curator5.Curator5ZookeeperTransporter

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/test/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperClientTest.java

@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.dubbo.remoting.zookeeper.curator5;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.remoting.zookeeper.ChildListener;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.test.TestingServer;
+import org.apache.zookeeper.WatchedEvent;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+
+public class Curator5ZookeeperClientTest {
+    private TestingServer zkServer;
+    private Curator5ZookeeperClient curatorClient;
+    CuratorFramework client = null;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        int zkServerPort = NetUtils.getAvailablePort();
+        zkServer = new TestingServer(zkServerPort, true);
+        curatorClient = new Curator5ZookeeperClient(URL.valueOf("zookeeper://127.0.0.1:" +
+                zkServerPort + "/org.apache.dubbo.registry.RegistryService"));
+        client = CuratorFrameworkFactory.newClient(zkServer.getConnectString(), new ExponentialBackoffRetry(1000, 3));
+        client.start();
+    }
+
+    @Test
+    public void testCheckExists() {
+        String path = "/dubbo/org.apache.dubbo.demo.DemoService/providers";
+        curatorClient.create(path, false);
+        assertThat(curatorClient.checkExists(path), is(true));
+        assertThat(curatorClient.checkExists(path + "/noneexits"), is(false));
+    }
+
+    @Test
+    public void testChildrenPath() {
+        String path = "/dubbo/org.apache.dubbo.demo.DemoService/providers";
+        curatorClient.create(path, false);
+        curatorClient.create(path + "/provider1", false);
+        curatorClient.create(path + "/provider2", false);
+
+        List<String> children = curatorClient.getChildren(path);
+        assertThat(children.size(), is(2));
+    }
+
+    @Test
+    public void testChildrenListener() throws InterruptedException {
+        String path = "/dubbo/org.apache.dubbo.demo.DemoService/providers";
+        curatorClient.create(path, false);
+        final CountDownLatch countDownLatch = new CountDownLatch(1);
+        curatorClient.addTargetChildListener(path, new Curator5ZookeeperClient.CuratorWatcherImpl() {
+
+            @Override
+            public void process(WatchedEvent watchedEvent) throws Exception {
+                countDownLatch.countDown();
+            }
+        });
+        curatorClient.createPersistent(path + "/provider1");
+        countDownLatch.await();
+    }
+
+
+    @Test
+    public void testWithInvalidServer() {
+        Assertions.assertThrows(IllegalStateException.class, () -> {
+            curatorClient = new Curator5ZookeeperClient(URL.valueOf("zookeeper://127.0.0.1:1/service"));
+            curatorClient.create("/testPath", true);
+        });
+    }
+
+    @Test
+    public void testWithStoppedServer() throws IOException {
+        Assertions.assertThrows(IllegalStateException.class, () -> {
+            curatorClient.create("/testPath", true);
+            zkServer.stop();
+            curatorClient.delete("/testPath");
+        });
+    }
+
+    @Test
+    public void testRemoveChildrenListener() {
+        ChildListener childListener = mock(ChildListener.class);
+        curatorClient.addChildListener("/children", childListener);
+        curatorClient.removeChildListener("/children", childListener);
+    }
+
+    @Test
+    public void testCreateExistingPath() {
+        curatorClient.create("/pathOne", false);
+        curatorClient.create("/pathOne", false);
+    }
+
+    @Test
+    public void testConnectedStatus() {
+        curatorClient.createEphemeral("/testPath");
+        boolean connected = curatorClient.isConnected();
+        assertThat(connected, is(true));
+    }
+
+    @Test
+    public void testCreateContent4Persistent() {
+        String path = "/curatorTest4CrContent/content.data";
+        String content = "createContentTest";
+        curatorClient.delete(path);
+        assertThat(curatorClient.checkExists(path), is(false));
+        assertNull(curatorClient.getContent(path));
+
+        curatorClient.create(path, content, false);
+        assertThat(curatorClient.checkExists(path), is(true));
+        assertEquals(curatorClient.getContent(path), content);
+    }
+
+    @Test
+    public void testCreateContent4Temp() {
+        String path = "/curatorTest4CrContent/content.data";
+        String content = "createContentTest";
+        curatorClient.delete(path);
+        assertThat(curatorClient.checkExists(path), is(false));
+        assertNull(curatorClient.getContent(path));
+
+        curatorClient.create(path, content, true);
+        assertThat(curatorClient.checkExists(path), is(true));
+        assertEquals(curatorClient.getContent(path), content);
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        curatorClient.close();
+        zkServer.stop();
+    }
+
+    @Test
+    public void testAddTargetDataListener() throws Exception {
+        String listenerPath = "/dubbo/service.name/configuration";
+        String path = listenerPath + "/dat/data";
+        String value = "vav";
+
+        curatorClient.create(path + "/d.json", value, true);
+        String valueFromCache = curatorClient.getContent(path + "/d.json");
+        Assertions.assertEquals(value, valueFromCache);
+        final AtomicInteger atomicInteger = new AtomicInteger(0);
+        curatorClient.addTargetDataListener(path + "/d.json", new Curator5ZookeeperClient.NodeCacheListenerImpl() {
+
+            @Override
+            public void nodeChanged() throws Exception {
+                atomicInteger.incrementAndGet();
+            }
+        });
+
+        valueFromCache = curatorClient.getContent(path + "/d.json");
+        Assertions.assertNotNull(valueFromCache);
+
+        Thread.sleep(100);
+        curatorClient.getClient().setData().forPath(path + "/d.json", "foo".getBytes());
+        Thread.sleep(100);
+        curatorClient.getClient().setData().forPath(path + "/d.json", "bar".getBytes());
+        curatorClient.delete(path + "/d.json");
+        valueFromCache = curatorClient.getContent(path + "/d.json");
+        Assertions.assertNull(valueFromCache);
+        Thread.sleep(2000L);
+        Assertions.assertTrue(3L <= atomicInteger.get());
+    }
+}

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/test/java/org/apache/dubbo/remoting/zookeeper/curator5/Curator5ZookeeperTransporterTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.dubbo.remoting.zookeeper.curator5;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
+
+import org.apache.curator.test.TestingServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsNot.not;
+import static org.hamcrest.core.IsNull.nullValue;
+
+public class Curator5ZookeeperTransporterTest {
+    private TestingServer zkServer;
+    private ZookeeperClient zookeeperClient;
+    private Curator5ZookeeperTransporter curatorZookeeperTransporter;
+    private int zkServerPort;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        zkServerPort = NetUtils.getAvailablePort();
+        zkServer = new TestingServer(zkServerPort, true);
+        zookeeperClient = new Curator5ZookeeperTransporter().connect(URL.valueOf("zookeeper://127.0.0.1:" +
+                zkServerPort + "/service"));
+        curatorZookeeperTransporter = new Curator5ZookeeperTransporter();
+    }
+
+    @Test
+    public void testZookeeperClient() {
+        assertThat(zookeeperClient, not(nullValue()));
+        zookeeperClient.close();
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        zkServer.stop();
+    }
+}

dubbo-remoting/dubbo-remoting-zookeeper-curator5/src/test/java/org/apache/dubbo/remoting/zookeeper/curator5/support/AbstractZookeeperTransporterTest.java

@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.dubbo.remoting.zookeeper.curator5.support;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.remoting.zookeeper.AbstractZookeeperTransporter;
+import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
+import org.apache.dubbo.remoting.zookeeper.curator5.Curator5ZookeeperTransporter;
+
+import org.apache.curator.test.TestingServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsNot.not;
+import static org.hamcrest.core.IsNull.nullValue;
+
+/**
+ * AbstractZookeeperTransporterTest
+ */
+public class AbstractZookeeperTransporterTest {
+    private TestingServer zkServer;
+    private ZookeeperClient zookeeperClient;
+    private AbstractZookeeperTransporter abstractZookeeperTransporter;
+    private int zkServerPort;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        zkServerPort = NetUtils.getAvailablePort();
+        zkServer = new TestingServer(zkServerPort, true);
+        zookeeperClient = new Curator5ZookeeperTransporter().connect(URL.valueOf("zookeeper://127.0.0.1:" +
+                zkServerPort + "/service"));
+        abstractZookeeperTransporter = new Curator5ZookeeperTransporter();
+    }
+
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        zkServer.stop();
+    }
+
+    @Test
+    public void testZookeeperClient() {
+        assertThat(zookeeperClient, not(nullValue()));
+        zookeeperClient.close();
+    }
+
+    @Test
+    public void testGetURLBackupAddress() {
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?backup=127.0.0.1:" + 9099 + "&application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        List<String> stringList = abstractZookeeperTransporter.getURLBackupAddress(url);
+        Assertions.assertEquals(stringList.size(), 2);
+        Assertions.assertEquals(stringList.get(0), "127.0.0.1:" + zkServerPort);
+        Assertions.assertEquals(stringList.get(1), "127.0.0.1:9099");
+    }
+
+    @Test
+    public void testGetURLBackupAddressNoBack() {
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        List<String> stringList = abstractZookeeperTransporter.getURLBackupAddress(url);
+        Assertions.assertEquals(stringList.size(), 1);
+        Assertions.assertEquals(stringList.get(0), "127.0.0.1:" + zkServerPort);
+    }
+
+    @Test
+    public void testFetchAndUpdateZookeeperClientCache() throws Exception {
+        int zkServerPort2 = NetUtils.getAvailablePort();
+        TestingServer zkServer2 = new TestingServer(zkServerPort2, true);
+
+        int zkServerPort3 = NetUtils.getAvailablePort();
+        TestingServer zkServer3 = new TestingServer(zkServerPort3, true);
+
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?backup=127.0.0.1:" + zkServerPort3 + ",127.0.0.1:" + zkServerPort2 + "&application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        ZookeeperClient newZookeeperClient = abstractZookeeperTransporter.connect(url);
+        //just for connected
+        newZookeeperClient.getContent("/dubbo/test");
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 3);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort), newZookeeperClient);
+
+        URL url2 = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.metadata.store.MetadataReport?address=zookeeper://127.0.0.1:2181&application=metadatareport-local-xml-provider2&cycle-report=false&interface=org.apache.dubbo.metadata.store.MetadataReport&retry-period=4590&retry-times=23&sync-report=true");
+        checkFetchAndUpdateCacheNotNull(url2);
+        URL url3 = URL.valueOf("zookeeper://127.0.0.1:8778/org.apache.dubbo.metadata.store.MetadataReport?backup=127.0.0.1:" + zkServerPort3 + "&address=zookeeper://127.0.0.1:2181&application=metadatareport-local-xml-provider2&cycle-report=false&interface=org.apache.dubbo.metadata.store.MetadataReport&retry-period=4590&retry-times=23&sync-report=true");
+        checkFetchAndUpdateCacheNotNull(url3);
+
+        zkServer2.stop();
+        zkServer3.stop();
+    }
+
+    private void checkFetchAndUpdateCacheNotNull(URL url) {
+        List<String> addressList = abstractZookeeperTransporter.getURLBackupAddress(url);
+        ZookeeperClient zookeeperClient = abstractZookeeperTransporter.fetchAndUpdateZookeeperClientCache(addressList);
+        Assertions.assertNotNull(zookeeperClient);
+    }
+
+    @Test
+    public void testRepeatConnect() {
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        URL url2 = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.metadata.store.MetadataReport?address=zookeeper://127.0.0.1:2181&application=metadatareport-local-xml-provider2&cycle-report=false&interface=org.apache.dubbo.metadata.store.MetadataReport&retry-period=4590&retry-times=23&sync-report=true");
+        ZookeeperClient newZookeeperClient = abstractZookeeperTransporter.connect(url);
+        //just for connected
+        newZookeeperClient.getContent("/dubbo/test");
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 1);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort), newZookeeperClient);
+        Assertions.assertTrue(newZookeeperClient.isConnected());
+
+        ZookeeperClient newZookeeperClient2 = abstractZookeeperTransporter.connect(url2);
+        //just for connected
+        newZookeeperClient2.getContent("/dubbo/test");
+        Assertions.assertEquals(newZookeeperClient, newZookeeperClient2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 1);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort), newZookeeperClient);
+    }
+
+    @Test
+    public void testNotRepeatConnect() throws Exception {
+        int zkServerPort2 = NetUtils.getAvailablePort();
+        TestingServer zkServer2 = new TestingServer(zkServerPort2, true);
+
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        URL url2 = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort2 + "/org.apache.dubbo.metadata.store.MetadataReport?address=zookeeper://127.0.0.1:2181&application=metadatareport-local-xml-provider2&cycle-report=false&interface=org.apache.dubbo.metadata.store.MetadataReport&retry-period=4590&retry-times=23&sync-report=true");
+        ZookeeperClient newZookeeperClient = abstractZookeeperTransporter.connect(url);
+        //just for connected
+        newZookeeperClient.getContent("/dubbo/test");
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 1);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort), newZookeeperClient);
+
+        ZookeeperClient newZookeeperClient2 = abstractZookeeperTransporter.connect(url2);
+        //just for connected
+        newZookeeperClient2.getContent("/dubbo/test");
+        Assertions.assertNotEquals(newZookeeperClient, newZookeeperClient2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort2), newZookeeperClient2);
+
+        zkServer2.stop();
+    }
+
+    @Test
+    public void testRepeatConnectForBackUpAdd() throws Exception {
+        int zkServerPort2 = NetUtils.getAvailablePort();
+        TestingServer zkServer2 = new TestingServer(zkServerPort2, true);
+
+        int zkServerPort3 = NetUtils.getAvailablePort();
+        TestingServer zkServer3 = new TestingServer(zkServerPort3, true);
+
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?backup=127.0.0.1:" + zkServerPort2 + "&application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        URL url2 = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort2 + "/org.apache.dubbo.metadata.store.MetadataReport?backup=127.0.0.1:" + zkServerPort3 + "&address=zookeeper://127.0.0.1:2181&application=metadatareport-local-xml-provider2&cycle-report=false&interface=org.apache.dubbo.metadata.store.MetadataReport&retry-period=4590&retry-times=23&sync-report=true");
+        ZookeeperClient newZookeeperClient = abstractZookeeperTransporter.connect(url);
+        //just for connected
+        newZookeeperClient.getContent("/dubbo/test");
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort), newZookeeperClient);
+
+        ZookeeperClient newZookeeperClient2 = abstractZookeeperTransporter.connect(url2);
+        //just for connected
+        newZookeeperClient2.getContent("/dubbo/test");
+        Assertions.assertEquals(newZookeeperClient, newZookeeperClient2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 3);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort2), newZookeeperClient2);
+
+        zkServer2.stop();
+        zkServer3.stop();
+    }
+
+    @Test
+    public void testRepeatConnectForNoMatchBackUpAdd() throws Exception {
+        int zkServerPort2 = NetUtils.getAvailablePort();
+        TestingServer zkServer2 = new TestingServer(zkServerPort2, true);
+
+        int zkServerPort3 = NetUtils.getAvailablePort();
+        TestingServer zkServer3 = new TestingServer(zkServerPort3, true);
+
+        URL url = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort + "/org.apache.dubbo.registry.RegistryService?backup=127.0.0.1:" + zkServerPort3 + "&application=metadatareport-local-xml-provider2&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=47418&specVersion=2.7.0-SNAPSHOT&timestamp=1547102428828");
+        URL url2 = URL.valueOf("zookeeper://127.0.0.1:" + zkServerPort2 + "/org.apache.dubbo.metadata.store.MetadataReport?address=zookeeper://127.0.0.1:2181&application=metadatareport-local-xml-provider2&cycle-report=false&interface=org.apache.dubbo.metadata.store.MetadataReport&retry-period=4590&retry-times=23&sync-report=true");
+        ZookeeperClient newZookeeperClient = abstractZookeeperTransporter.connect(url);
+        //just for connected
+        newZookeeperClient.getContent("/dubbo/test");
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort), newZookeeperClient);
+
+        ZookeeperClient newZookeeperClient2 = abstractZookeeperTransporter.connect(url2);
+        //just for connected
+        newZookeeperClient2.getContent("/dubbo/test");
+        Assertions.assertNotEquals(newZookeeperClient, newZookeeperClient2);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().size(), 3);
+        Assertions.assertEquals(abstractZookeeperTransporter.getZookeeperClientMap().get("127.0.0.1:" + zkServerPort2), newZookeeperClient2);
+
+        zkServer2.stop();
+        zkServer3.stop();
+    }
+
+    @Test
+    public void testSameHostWithDifferentUser() throws Exception {
+        int zkPort1 = NetUtils.getAvailablePort();
+        int zkPort2 = NetUtils.getAvailablePort();
+        try (TestingServer zkServer1 = new TestingServer(zkPort1, true)) {
+            try (TestingServer zkServer2 = new TestingServer(zkPort2, true)) {
+                URL url1 = URL.valueOf("zookeeper://us1:pw1@127.0.0.1:" + zkPort1 + "/path1");
+                URL url2 = URL.valueOf("zookeeper://us2:pw2@127.0.0.1:" + zkPort1 + "/path2");
+
+                ZookeeperClient client1 = abstractZookeeperTransporter.connect(url1);
+                ZookeeperClient client2 = abstractZookeeperTransporter.connect(url2);
+
+                assertThat(client1, not(client2));
+            }
+        }
+    }
+}

dubbo-remoting/dubbo-remoting-zookeeper/src/main/java/org/apache/dubbo/remoting/zookeeper/curator/CuratorZookeeperClient.java

@@ -24,7 +24,7 @@ import org.apache.dubbo.remoting.zookeeper.ChildListener;
 import org.apache.dubbo.remoting.zookeeper.DataListener;
 import org.apache.dubbo.remoting.zookeeper.EventType;
 import org.apache.dubbo.remoting.zookeeper.StateListener;
-import org.apache.dubbo.remoting.zookeeper.support.AbstractZookeeperClient;
+import org.apache.dubbo.remoting.zookeeper.AbstractZookeeperClient;
 
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;

dubbo-remoting/dubbo-remoting-zookeeper/src/main/java/org/apache/dubbo/remoting/zookeeper/curator/CuratorZookeeperTransporter.java

@@ -18,7 +18,7 @@ package org.apache.dubbo.remoting.zookeeper.curator;
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
-import org.apache.dubbo.remoting.zookeeper.support.AbstractZookeeperTransporter;
+import org.apache.dubbo.remoting.zookeeper.AbstractZookeeperTransporter;
 
 public class CuratorZookeeperTransporter extends AbstractZookeeperTransporter {
     @Override

dubbo-remoting/dubbo-remoting-zookeeper/src/test/java/org/apache/dubbo/remoting/zookeeper/curator/support/AbstractZookeeperTransporterTest.java

@@ -14,10 +14,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.zookeeper.support;
+package org.apache.dubbo.remoting.zookeeper.curator.support;
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.remoting.zookeeper.AbstractZookeeperTransporter;
 import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
 import org.apache.dubbo.remoting.zookeeper.curator.CuratorZookeeperTransporter;
 

dubbo-remoting/pom.xml

@@ -35,6 +35,7 @@
         <module>dubbo-remoting-http</module>
         <module>dubbo-remoting-netty</module>
         <module>dubbo-remoting-zookeeper</module>
+        <module>dubbo-remoting-zookeeper-curator5</module>
         <module>dubbo-remoting-netty4</module>
     </modules>
-</project>
+</project>