CLUSTER_INFO checking

src/main/java/org/redisson/connection/BaseLoadBalancer.java

@@ -28,7 +28,6 @@
 import org.redisson.client.RedisConnectionException;
 import org.redisson.client.RedisException;
 import org.redisson.client.RedisPubSubConnection;
-import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.RedisCommands;
 import org.redisson.misc.ReclosableLatch;
 import org.slf4j.Logger;
@@ -38,15 +37,12 @@ abstract class BaseLoadBalancer implements LoadBalancer {
 
     private final Logger log = LoggerFactory.getLogger(getClass());
 
-    private Codec codec;
-
     private MasterSlaveServersConfig config;
 
     private final ReclosableLatch clientsEmpty = new ReclosableLatch();
     final Queue<SubscribesConnectionEntry> clients = new ConcurrentLinkedQueue<SubscribesConnectionEntry>();
 
-    public void init(Codec codec, MasterSlaveServersConfig config) {
-        this.codec = codec;
+    public void init(MasterSlaveServersConfig config) {
         this.config = config;
     }
 
@@ -227,4 +223,10 @@ public void shutdown() {
         }
     }
 
+    public void shutdownAsync() {
+        for (SubscribesConnectionEntry entry : clients) {
+            entry.getClient().shutdownAsync();
+        }
+    }
+
 }

src/main/java/org/redisson/connection/ClusterConnectionManager.java

@@ -62,7 +62,7 @@ public ClusterConnectionManager(ClusterServersConfig cfg, Config config) {
                 RedisConnection connection = client.connect();
                 String nodesValue = connection.sync(RedisCommands.CLUSTER_NODES);
 
-                Map<Integer, ClusterPartition> partitions = extractPartitions(nodesValue);
+                Map<Integer, ClusterPartition> partitions = parsePartitions(nodesValue);
                 for (ClusterPartition partition : partitions.values()) {
                     addMasterEntry(partition, cfg);
                 }
@@ -85,14 +85,28 @@ protected void initEntry(MasterSlaveServersConfig config) {
 
     private void addMasterEntry(ClusterPartition partition, ClusterServersConfig cfg) {
         if (partition.isMasterFail()) {
+            log.warn("master: {} for slot range: {}-{} add failed. Reason - server has FAIL flag", partition.getMasterAddress(), partition.getStartSlot(), partition.getEndSlot());
             return;
         }
 
-        MasterSlaveServersConfig c = create(cfg);
+        RedisClient client = createClient(partition.getMasterAddress().getHost(), partition.getMasterAddress().getPort(), cfg.getTimeout());
+        try {
+            RedisConnection c = client.connect();
+            String info = c.sync(RedisCommands.CLUSTER_INFO);
+            Map<String, String> params = parseInfo(info);
+            if ("fail".equals(params.get("cluster_state"))) {
+                log.warn("master: {} for slot range: {}-{} add failed. Reason - cluster_state:fail", partition.getMasterAddress(), partition.getStartSlot(), partition.getEndSlot());
+                return;
+            }
+        } finally {
+            client.shutdownAsync();
+        }
+
+        MasterSlaveServersConfig config = create(cfg);
         log.info("master: {} for slot range: {}-{} added", partition.getMasterAddress(), partition.getStartSlot(), partition.getEndSlot());
-        c.setMasterAddress(partition.getMasterAddress());
+        config.setMasterAddress(partition.getMasterAddress());
 
-        SingleEntry entry = new SingleEntry(codec, this, c);
+        SingleEntry entry = new SingleEntry(this, config);
         entries.put(partition.getEndSlot(), entry);
         lastPartitions.put(partition.getEndSlot(), partition);
     }
@@ -110,7 +124,7 @@ public void run() {
 
                             log.debug("cluster nodes state: {}", nodesValue);
 
-                            Map<Integer, ClusterPartition> partitions = extractPartitions(nodesValue);
+                            Map<Integer, ClusterPartition> partitions = parsePartitions(nodesValue);
                             for (ClusterPartition newPart : partitions.values()) {
                                 for (ClusterPartition part : lastPartitions.values()) {
                                     if (newPart.getMasterAddress().equals(part.getMasterAddress())) {
@@ -187,7 +201,16 @@ private void checkSlotsChange(ClusterServersConfig cfg, Map<Integer, ClusterPart
         }
     }
 
-    private Map<Integer, ClusterPartition> extractPartitions(String nodesValue) {
+    private Map<String, String> parseInfo(String value) {
+        Map<String, String> result = new HashMap<String, String>();
+        for (String entry : value.split("\r\n|\n")) {
+            String[] parts = entry.split(":");
+            result.put(parts[0], parts[1]);
+        }
+        return result;
+    }
+
+    private Map<Integer, ClusterPartition> parsePartitions(String nodesValue) {
         Map<String, ClusterPartition> partitions = new HashMap<String, ClusterPartition>();
         Map<Integer, ClusterPartition> result = new HashMap<Integer, ClusterPartition>();
         List<ClusterNodeInfo> nodes = parse(nodesValue);

src/main/java/org/redisson/connection/LoadBalancer.java

@@ -24,13 +24,15 @@
 
 public interface LoadBalancer {
 
+    void shutdownAsync();
+
     void shutdown();
 
     void unfreeze(String host, int port);
 
     Collection<RedisPubSubConnection> freeze(String host, int port);
 
-    void init(Codec codec, MasterSlaveServersConfig config);
+    void init(MasterSlaveServersConfig config);
 
     void add(SubscribesConnectionEntry entry);
 

src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java

@@ -30,9 +30,11 @@
 import org.redisson.client.BaseRedisPubSubListener;
 import org.redisson.client.RedisClient;
 import org.redisson.client.RedisConnection;
+import org.redisson.client.RedisException;
 import org.redisson.client.RedisPubSubConnection;
 import org.redisson.client.RedisPubSubListener;
 import org.redisson.client.codec.Codec;
+import org.redisson.client.protocol.RedisCommands;
 import org.redisson.client.protocol.pubsub.PubSubType;
 import org.redisson.misc.InfinitySemaphoreLatch;
 import org.slf4j.Logger;
@@ -122,14 +124,10 @@ protected void init(MasterSlaveServersConfig config) {
     }
 
     protected void initEntry(MasterSlaveServersConfig config) {
-        MasterSlaveEntry entry = new MasterSlaveEntry(codec, this, config);
+        MasterSlaveEntry entry = new MasterSlaveEntry(this, config);
         entries.put(Integer.MAX_VALUE, entry);
     }
 
-    public static void main(String[] args) {
-        System.out.println(1210 % 100);
-    }
-
     protected void init(Config cfg) {
         if (cfg.isUseLinuxNativeEpoll()) {
             this.group = new EpollEventLoopGroup(cfg.getThreads());
@@ -141,14 +139,17 @@ protected void init(Config cfg) {
         this.codec = cfg.getCodec();
     }
 
+    @Override
     public RedisClient createClient(String host, int port) {
         return createClient(host, port, config.getTimeout());
     }
 
+    @Override
     public RedisClient createClient(String host, int port, int timeout) {
         return new RedisClient(group, socketChannelClass, host, port, timeout);
     }
 
+    @Override
     public <T> FutureListener<T> createReleaseWriteListener(final int slot,
                                     final RedisConnection conn, final Timeout timeout) {
         return new FutureListener<T>() {
@@ -161,6 +162,7 @@ public void operationComplete(io.netty.util.concurrent.Future<T> future) throws
         };
     }
 
+    @Override
     public <T> FutureListener<T> createReleaseReadListener(final int slot,
                                     final RedisConnection conn, final Timeout timeout) {
         return new FutureListener<T>() {

src/main/java/org/redisson/connection/MasterSlaveEntry.java

@@ -23,11 +23,8 @@
 import org.redisson.MasterSlaveServersConfig;
 import org.redisson.client.RedisClient;
 import org.redisson.client.RedisConnection;
-import org.redisson.client.RedisConnectionException;
 import org.redisson.client.RedisException;
 import org.redisson.client.RedisPubSubConnection;
-import org.redisson.client.codec.Codec;
-import org.redisson.client.protocol.RedisCommands;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,25 +42,22 @@ public class MasterSlaveEntry {
     volatile ConnectionEntry masterEntry;
 
     final MasterSlaveServersConfig config;
-    final Codec codec;
     final ConnectionManager connectionManager;
 
-    public MasterSlaveEntry(Codec codec, ConnectionManager connectionManager, MasterSlaveServersConfig config) {
-        this.codec = codec;
+    public MasterSlaveEntry(ConnectionManager connectionManager, MasterSlaveServersConfig config) {
         this.connectionManager = connectionManager;
         this.config = config;
 
         slaveBalancer = config.getLoadBalancer();
-        slaveBalancer.init(codec, config);
+        slaveBalancer.init(config);
 
         List<URI> addresses = new ArrayList<URI>(config.getSlaveAddresses());
         addresses.add(config.getMasterAddress());
         for (URI address : addresses) {
             RedisClient client = connectionManager.createClient(address.getHost(), address.getPort());
-            SubscribesConnectionEntry entry = new SubscribesConnectionEntry(client,
-                    config.getSlaveConnectionPoolSize(),
-                    config.getSlaveSubscriptionConnectionPoolSize());
-            slaveBalancer.add(entry);
+            slaveBalancer.add(new SubscribesConnectionEntry(client,
+                    this.config.getSlaveConnectionPoolSize(),
+                    this.config.getSlaveSubscriptionConnectionPoolSize()));
         }
         if (config.getSlaveAddresses().size() > 1) {
             slaveDown(config.getMasterAddress().getHost(), config.getMasterAddress().getPort());
@@ -73,8 +67,8 @@ public MasterSlaveEntry(Codec codec, ConnectionManager connectionManager, Master
     }
 
     public void setupMasterEntry(String host, int port) {
-        RedisClient masterClient = connectionManager.createClient(host, port);
-        masterEntry = new ConnectionEntry(masterClient, config.getMasterConnectionPoolSize());
+        RedisClient client = connectionManager.createClient(host, port);
+        masterEntry = new ConnectionEntry(client, config.getMasterConnectionPoolSize());
     }
 
     public Collection<RedisPubSubConnection> slaveDown(String host, int port) {
@@ -113,7 +107,7 @@ public void changeMaster(String host, int port) {
 
     public void shutdownMasterAsync() {
         masterEntry.getClient().shutdownAsync();
-        slaveBalancer.shutdown();
+        slaveBalancer.shutdownAsync();
     }
 
     public RedisConnection connectionWriteOp() {

src/main/java/org/redisson/connection/SingleConnectionManager.java

@@ -40,7 +40,7 @@ public SingleConnectionManager(SingleServerConfig cfg, Config config) {
 
     @Override
     protected void initEntry(MasterSlaveServersConfig config) {
-        SingleEntry entry = new SingleEntry(codec, this, config);
+        SingleEntry entry = new SingleEntry(this, config);
         entries.put(Integer.MAX_VALUE, entry);
     }
 

src/main/java/org/redisson/connection/SingleEntry.java

@@ -20,13 +20,12 @@
 import org.redisson.client.RedisConnection;
 import org.redisson.client.RedisConnectionException;
 import org.redisson.client.RedisPubSubConnection;
-import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.RedisCommands;
 
 public class SingleEntry extends MasterSlaveEntry {
 
-    public SingleEntry(Codec codec, ConnectionManager connectionManager, MasterSlaveServersConfig config) {
-        super(codec, connectionManager, config);
+    public SingleEntry(ConnectionManager connectionManager, MasterSlaveServersConfig config) {
+        super(connectionManager, config);
     }
 
     @Override