Merge branch 'redis:master' into master

redis · Aug 14, 2024 · 3aa572a · 3aa572a
2 parents 288617c + cd9a1ab
commit 3aa572a
Show file tree

Hide file tree

Showing 10 changed files with 207 additions and 6 deletions.
diff --git a/src/main/java/redis/clients/jedis/Connection.java b/src/main/java/redis/clients/jedis/Connection.java
@@ -5,6 +5,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.net.Socket;
+import java.net.SocketAddress;
 import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;
@@ -37,6 +38,8 @@ public class Connection implements Closeable {
   private int soTimeout = 0;
   private int infiniteSoTimeout = 0;
   private boolean broken = false;
+  private boolean strValActive;
+  private String strVal;
 
   public Connection() {
     this(Protocol.DEFAULT_HOST, Protocol.DEFAULT_PORT);
@@ -72,6 +75,63 @@ public String toString() {
     return "Connection{" + socketFactory + "}";
   }
 
+  public String toIdentityString() {
+    if (strValActive == broken && strVal != null) {
+      return strVal;
+    }
+
+    int id = hashCode();
+    String classInfo = getClass().toString();
+
+    if (socket == null) {
+      StringBuilder buf = new StringBuilder(56)
+          .append("[")
+          .append(classInfo)
+          .append(", id: 0x")
+          .append(id)
+          .append(']');
+      return buf.toString();
+    }
+
+    SocketAddress remoteAddr = socket.getRemoteSocketAddress();
+    SocketAddress localAddr = socket.getLocalSocketAddress();
+    if (remoteAddr != null) {
+      StringBuilder buf = new StringBuilder(101)
+          .append("[")
+          .append(classInfo)
+          .append(", id: 0x")
+          .append(id)
+          .append(", L:")
+          .append(localAddr)
+          .append(broken? " ! " : " - ")
+          .append("R:")
+          .append(remoteAddr)
+          .append(']');
+      strVal = buf.toString();
+    } else if (localAddr != null) {
+      StringBuilder buf = new StringBuilder(64)
+          .append("[")
+          .append(classInfo)
+          .append(", id: 0x")
+          .append(id)
+          .append(", L:")
+          .append(localAddr)
+          .append(']');
+      strVal = buf.toString();
+    } else {
+      StringBuilder buf = new StringBuilder(56)
+          .append("[")
+          .append(classInfo)
+          .append(", id: 0x")
+          .append(id)
+          .append(']');
+      strVal = buf.toString();
+    }
+
+    strValActive = broken;
+    return strVal;
+  }
+
   public final RedisProtocol getRedisProtocol() {
     return protocol;
   }
@@ -435,6 +495,11 @@ private void initializeFromClientConfig(final JedisClientConfig config) {
         }
       }
 
+      // set readonly flag to ALL connections (including master nodes) when enable read from replica
+      if (config.isReadOnlyForReplica()) {
+        fireAndForgetMsg.add(new CommandArguments(Command.READONLY));
+      }
+
       for (CommandArguments arg : fireAndForgetMsg) {
         sendCommand(arg);
       }

diff --git a/src/main/java/redis/clients/jedis/DefaultJedisClientConfig.java b/src/main/java/redis/clients/jedis/DefaultJedisClientConfig.java
@@ -26,11 +26,13 @@ public final class DefaultJedisClientConfig implements JedisClientConfig {
 
   private final ClientSetInfoConfig clientSetInfoConfig;
 
+  private final boolean readOnlyForReplica;
+
   private DefaultJedisClientConfig(RedisProtocol protocol, int connectionTimeoutMillis, int soTimeoutMillis,
       int blockingSocketTimeoutMillis, Supplier<RedisCredentials> credentialsProvider, int database,
       String clientName, boolean ssl, SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,
       HostnameVerifier hostnameVerifier, HostAndPortMapper hostAndPortMapper,
-      ClientSetInfoConfig clientSetInfoConfig) {
+      ClientSetInfoConfig clientSetInfoConfig, boolean readOnlyForReplica) {
     this.redisProtocol = protocol;
     this.connectionTimeoutMillis = connectionTimeoutMillis;
     this.socketTimeoutMillis = soTimeoutMillis;
@@ -44,6 +46,7 @@ private DefaultJedisClientConfig(RedisProtocol protocol, int connectionTimeoutMi
     this.hostnameVerifier = hostnameVerifier;
     this.hostAndPortMapper = hostAndPortMapper;
     this.clientSetInfoConfig = clientSetInfoConfig;
+    this.readOnlyForReplica = readOnlyForReplica;
   }
 
   @Override
@@ -122,6 +125,11 @@ public ClientSetInfoConfig getClientSetInfoConfig() {
     return clientSetInfoConfig;
   }
 
+  @Override
+  public boolean isReadOnlyForReplica() {
+    return readOnlyForReplica;
+  }
+
   public static Builder builder() {
     return new Builder();
   }
@@ -149,6 +157,8 @@ public static class Builder {
 
     private ClientSetInfoConfig clientSetInfoConfig = ClientSetInfoConfig.DEFAULT;
 
+    private boolean readOnlyForReplicas = false;
+
     private Builder() {
     }
 
@@ -160,7 +170,8 @@ public DefaultJedisClientConfig build() {
 
       return new DefaultJedisClientConfig(redisProtocol, connectionTimeoutMillis, socketTimeoutMillis,
           blockingSocketTimeoutMillis, credentialsProvider, database, clientName, ssl,
-          sslSocketFactory, sslParameters, hostnameVerifier, hostAndPortMapper, clientSetInfoConfig);
+          sslSocketFactory, sslParameters, hostnameVerifier, hostAndPortMapper, clientSetInfoConfig,
+          readOnlyForReplicas);
     }
 
     /**
@@ -255,6 +266,11 @@ public Builder clientSetInfoConfig(ClientSetInfoConfig setInfoConfig) {
       this.clientSetInfoConfig = setInfoConfig;
       return this;
     }
+
+    public Builder readOnlyForReplicas() {
+      this.readOnlyForReplicas = true;
+      return this;
+    }
   }
 
   public static DefaultJedisClientConfig create(int connectionTimeoutMillis, int soTimeoutMillis,
@@ -264,7 +280,8 @@ public static DefaultJedisClientConfig create(int connectionTimeoutMillis, int s
     return new DefaultJedisClientConfig(null,
         connectionTimeoutMillis, soTimeoutMillis, blockingSocketTimeoutMillis,
         new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(user, password)), database,
-        clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier, hostAndPortMapper, null);
+        clientName, ssl, sslSocketFactory, sslParameters, hostnameVerifier, hostAndPortMapper, null,
+        false);
   }
 
   public static DefaultJedisClientConfig copyConfig(JedisClientConfig copy) {
@@ -273,6 +290,6 @@ public static DefaultJedisClientConfig copyConfig(JedisClientConfig copy) {
         copy.getBlockingSocketTimeoutMillis(), copy.getCredentialsProvider(),
         copy.getDatabase(), copy.getClientName(), copy.isSsl(), copy.getSslSocketFactory(),
         copy.getSslParameters(), copy.getHostnameVerifier(), copy.getHostAndPortMapper(),
-        copy.getClientSetInfoConfig());
+        copy.getClientSetInfoConfig(), copy.isReadOnlyForReplica());
   }
 }
diff --git a/src/main/java/redis/clients/jedis/Jedis.java b/src/main/java/redis/clients/jedis/Jedis.java
@@ -3431,7 +3431,7 @@ public String shutdownAbort() {
    * All the fields are in the form field:value
    *
    * <pre>
-   * edis_version:0.07
+   * redis_version:0.07
    * connected_clients:1
    * connected_slaves:0
    * used_memory:3187

diff --git a/src/main/java/redis/clients/jedis/JedisClientConfig.java b/src/main/java/redis/clients/jedis/JedisClientConfig.java
@@ -80,6 +80,10 @@ default HostAndPortMapper getHostAndPortMapper() {
     return null;
   }
 
+  default boolean isReadOnlyForReplica() {
+    return false;
+  }
+
   /**
    * Modify the behavior of internally executing CLIENT SETINFO command.
    * @return CLIENT SETINFO config

diff --git a/src/main/java/redis/clients/jedis/JedisCluster.java b/src/main/java/redis/clients/jedis/JedisCluster.java
@@ -7,6 +7,7 @@
 
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 
+import redis.clients.jedis.executors.ClusterCommandExecutor;
 import redis.clients.jedis.providers.ClusterConnectionProvider;
 import redis.clients.jedis.util.JedisClusterCRC16;
 
@@ -266,4 +267,11 @@ public ClusterPipeline pipelined() {
   public AbstractTransaction transaction(boolean doMulti) {
     throw new UnsupportedOperationException();
   }
+
+  public final <T> T executeCommandToReplica(CommandObject<T> commandObject) {
+    if (!(executor instanceof ClusterCommandExecutor)) {
+      throw new UnsupportedOperationException("Support only execute to replica in ClusterCommandExecutor");
+    }
+    return ((ClusterCommandExecutor) executor).executeCommandToReplica(commandObject);
+  }
 }
diff --git a/src/main/java/redis/clients/jedis/JedisClusterInfoCache.java b/src/main/java/redis/clients/jedis/JedisClusterInfoCache.java
@@ -38,6 +38,7 @@ public class JedisClusterInfoCache {
   private final Map<String, ConnectionPool> nodes = new HashMap<>();
   private final ConnectionPool[] slots = new ConnectionPool[Protocol.CLUSTER_HASHSLOTS];
   private final HostAndPort[] slotNodes = new HostAndPort[Protocol.CLUSTER_HASHSLOTS];
+  private final List<ConnectionPool>[] replicaSlots;
 
   private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
   private final Lock r = rwl.readLock();
@@ -85,6 +86,11 @@ public JedisClusterInfoCache(final JedisClientConfig clientConfig,
       topologyRefreshExecutor.scheduleWithFixedDelay(new TopologyRefreshTask(), topologyRefreshPeriod.toMillis(),
           topologyRefreshPeriod.toMillis(), TimeUnit.MILLISECONDS);
     }
+    if (clientConfig.isReadOnlyForReplica()) {
+      replicaSlots = new ArrayList[Protocol.CLUSTER_HASHSLOTS];
+    } else {
+      replicaSlots = null;
+    }
   }
 
   /**
@@ -144,6 +150,8 @@ public void discoverClusterNodesAndSlots(Connection jedis) {
           setupNodeIfNotExist(targetNode);
           if (i == MASTER_NODE_INDEX) {
             assignSlotsToNode(slotNums, targetNode);
+          } else if (clientConfig.isReadOnlyForReplica()) {
+            assignSlotsToReplicaNode(slotNums, targetNode);
           }
         }
       }
@@ -236,6 +244,8 @@ private void discoverClusterSlots(Connection jedis) {
           setupNodeIfNotExist(targetNode);
           if (i == MASTER_NODE_INDEX) {
             assignSlotsToNode(slotNums, targetNode);
+          } else if (clientConfig.isReadOnlyForReplica()) {
+            assignSlotsToReplicaNode(slotNums, targetNode);
           }
         }
       }
@@ -307,6 +317,21 @@ public void assignSlotsToNode(List<Integer> targetSlots, HostAndPort targetNode)
     }
   }
 
+  public void assignSlotsToReplicaNode(List<Integer> targetSlots, HostAndPort targetNode) {
+    w.lock();
+    try {
+      ConnectionPool targetPool = setupNodeIfNotExist(targetNode);
+      for (Integer slot : targetSlots) {
+        if (replicaSlots[slot] == null) {
+          replicaSlots[slot] = new ArrayList<>();
+        }
+        replicaSlots[slot].add(targetPool);
+      }
+    } finally {
+      w.unlock();
+    }
+  }
+
   public ConnectionPool getNode(String nodeKey) {
     r.lock();
     try {
@@ -338,6 +363,15 @@ public HostAndPort getSlotNode(int slot) {
     }
   }
 
+  public List<ConnectionPool> getSlotReplicaPools(int slot) {
+    r.lock();
+    try {
+      return replicaSlots[slot];
+    } finally {
+      r.unlock();
+    }
+  }
+
   public Map<String, ConnectionPool> getNodes() {
     r.lock();
     try {

diff --git a/src/main/java/redis/clients/jedis/executors/ClusterCommandExecutor.java b/src/main/java/redis/clients/jedis/executors/ClusterCommandExecutor.java
@@ -73,6 +73,14 @@ public final <T> T broadcastCommand(CommandObject<T> commandObject) {
 
   @Override
   public final <T> T executeCommand(CommandObject<T> commandObject) {
+    return doExecuteCommand(commandObject, false);
+  }
+
+  public final <T> T executeCommandToReplica(CommandObject<T> commandObject) {
+    return doExecuteCommand(commandObject, true);
+  }
+
+  private <T> T doExecuteCommand(CommandObject<T> commandObject, boolean toReplica) {
     Instant deadline = Instant.now().plus(maxTotalRetriesDuration);
 
     JedisRedirectionException redirect = null;
@@ -88,7 +96,8 @@ public final <T> T executeCommand(CommandObject<T> commandObject) {
             connection.executeCommand(Protocol.Command.ASKING);
           }
         } else {
-          connection = provider.getConnection(commandObject.getArguments());
+          connection = toReplica ? provider.getReplicaConnection(commandObject.getArguments())
+              : provider.getConnection(commandObject.getArguments());
         }
 
         return execute(connection, commandObject);

diff --git a/src/main/java/redis/clients/jedis/providers/ClusterConnectionProvider.java b/src/main/java/redis/clients/jedis/providers/ClusterConnectionProvider.java
@@ -6,6 +6,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 
 import redis.clients.jedis.ClusterCommandArguments;
@@ -102,6 +104,11 @@ public Connection getConnection(CommandArguments args) {
     return slot >= 0 ? getConnectionFromSlot(slot) : getConnection();
   }
 
+  public Connection getReplicaConnection(CommandArguments args) {
+    final int slot = ((ClusterCommandArguments) args).getCommandHashSlot();
+    return slot >= 0 ? getReplicaConnectionFromSlot(slot) : getConnection();
+  }
+
   @Override
   public Connection getConnection() {
     // In antirez's redis-rb-cluster implementation, getRandomConnection always
@@ -158,6 +165,25 @@ public Connection getConnectionFromSlot(int slot) {
     }
   }
 
+  public Connection getReplicaConnectionFromSlot(int slot) {
+    List<ConnectionPool> connectionPools = cache.getSlotReplicaPools(slot);
+    ThreadLocalRandom random = ThreadLocalRandom.current();
+    if (connectionPools != null && !connectionPools.isEmpty()) {
+      // pick up randomly a connection
+      int idx = random.nextInt(connectionPools.size());
+      return connectionPools.get(idx).getResource();
+    }
+
+    renewSlotCache();
+    connectionPools = cache.getSlotReplicaPools(slot);
+    if (connectionPools != null && !connectionPools.isEmpty()) {
+      int idx = random.nextInt(connectionPools.size());
+      return connectionPools.get(idx).getResource();
+    }
+
+    return getConnectionFromSlot(slot);
+  }
+
   @Override
   public Map<String, ConnectionPool> getConnectionMap() {
     return Collections.unmodifiableMap(getNodes());

diff --git a/src/test/java/redis/clients/jedis/ConnectionTest.java b/src/test/java/redis/clients/jedis/ConnectionTest.java
@@ -1,6 +1,7 @@
 package redis.clients.jedis;
 
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Test;
 
 import redis.clients.jedis.exceptions.JedisConnectionException;
@@ -40,4 +41,14 @@ public void checkCloseable() {
     client.connect();
     client.close();
   }
+
+  @Test
+  public void checkIdentityString() {
+    client = new Connection("localhost", 6379);
+    Assert.assertFalse(client.toIdentityString().contains("6379"));
+    client.connect();
+    Assert.assertTrue(client.toIdentityString().contains("6379"));
+    client.close();
+    Assert.assertTrue(client.toIdentityString().contains("6379"));
+  }
 }