[#3817]Support Redis cluster in Metadata Report

dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/MetadataReportConfig.java

@@ -71,6 +71,11 @@ public class MetadataReportConfig extends AbstractConfig {
      */
     private Boolean syncReport;
 
+    /**
+     * cluster
+     */
+    private Boolean cluster;
+
     public MetadataReportConfig() {
     }
 
@@ -174,4 +179,12 @@ public String getGroup() {
     public void setGroup(String group) {
         this.group = group;
     }
+
+    public Boolean getCluster() {
+        return cluster;
+    }
+
+    public void setCluster(Boolean cluster) {
+        this.cluster = cluster;
+    }
 }

dubbo-config/dubbo-config-spring/src/main/resources/META-INF/compat/dubbo.xsd

@@ -632,6 +632,11 @@
                 <xsd:documentation><![CDATA[ Sync or Async report. ]]></xsd:documentation>
             </xsd:annotation>
         </xsd:attribute>
+        <xsd:attribute name="cluster" type="xsd:boolean" use="optional">
+            <xsd:annotation>
+                <xsd:documentation><![CDATA[ Need cluster support, default false. ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute>
     </xsd:complexType>
 
     <xsd:complexType name="configCenterType">

dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd

@@ -626,6 +626,11 @@
                 <xsd:documentation><![CDATA[ Sync or Async report. ]]></xsd:documentation>
             </xsd:annotation>
         </xsd:attribute>
+        <xsd:attribute name="cluster" type="xsd:boolean" use="optional">
+            <xsd:annotation>
+                <xsd:documentation><![CDATA[ Need cluster support, default false. ]]></xsd:documentation>
+            </xsd:annotation>
+        </xsd:attribute>
     </xsd:complexType>
 
     <xsd:complexType name="configCenterType">

dubbo-metadata-report/dubbo-metadata-report-redis/src/main/java/org/apache/dubbo/metadata/store/redis/RedisMetadataReport.java

@@ -16,28 +16,46 @@
  */
 package org.apache.dubbo.metadata.store.redis;
 
+import org.apache.dubbo.common.Constants;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.metadata.identifier.MetadataIdentifier;
 import org.apache.dubbo.metadata.support.AbstractMetadataReport;
 import org.apache.dubbo.rpc.RpcException;
+
+import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisCluster;
 import redis.clients.jedis.JedisPool;
 import redis.clients.jedis.JedisPoolConfig;
 
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
 /**
  * RedisMetadataReport
  */
 public class RedisMetadataReport extends AbstractMetadataReport {
 
     private final static Logger logger = LoggerFactory.getLogger(RedisMetadataReport.class);
 
-    final JedisPool pool;
+    JedisPool pool;
+    Set<HostAndPort> jedisClusterNodes;
+
 
     public RedisMetadataReport(URL url) {
         super(url);
-        pool = new JedisPool(new JedisPoolConfig(), url.getHost(), url.getPort());
+        if (url.getParameter(Constants.CLUSTER_KEY, false)) {
+            jedisClusterNodes = new HashSet<HostAndPort>();
+            List<URL> urls = url.getBackupUrls();
+            for (URL tmpUrl : urls) {
+                jedisClusterNodes.add(new HostAndPort(tmpUrl.getHost(), tmpUrl.getPort()));
+            }
+        } else {
+            pool = new JedisPool(new JedisPoolConfig(), url.getHost(), url.getPort());
+        }
     }
 
     @Override
@@ -51,6 +69,23 @@ protected void doStoreConsumerMetadata(MetadataIdentifier consumerMetadataIdenti
     }
 
     private void storeMetadata(MetadataIdentifier metadataIdentifier, String v) {
+        if (pool != null) {
+            storeMetadataStandalone(metadataIdentifier, v);
+        } else {
+            storeMetadataInCluster(metadataIdentifier, v);
+        }
+    }
+
+    private void storeMetadataInCluster(MetadataIdentifier metadataIdentifier, String v) {
+        try (JedisCluster jedisCluster = new JedisCluster(jedisClusterNodes)) {
+            jedisCluster.set(metadataIdentifier.getIdentifierKey() + META_DATA_STORE_TAG, v);
+        } catch (Throwable e) {
+            logger.error("Failed to put " + metadataIdentifier + " to redis cluster " + v + ", cause: " + e.getMessage(), e);
+            throw new RpcException("Failed to put " + metadataIdentifier + " to redis cluster " + v + ", cause: " + e.getMessage(), e);
+        }
+    }
+
+    private void storeMetadataStandalone(MetadataIdentifier metadataIdentifier, String v) {
         try (Jedis jedis = pool.getResource()) {
             jedis.set(metadataIdentifier.getIdentifierKey() + META_DATA_STORE_TAG, v);
         } catch (Throwable e) {
@@ -59,5 +94,4 @@ private void storeMetadata(MetadataIdentifier metadataIdentifier, String v) {
         }
     }
 
-
 }