Upgrade to Jedis 4.2.

The upgrade also allows to enable support for GEOSEARCH, GEOSEARCHSTORE, XCLAIMJUSTID and XPENDING.

Closes: #2212
Original Pull Request: #2287

Author: mp911de

pom.xml

@@ -24,7 +24,7 @@
 		<xstream>1.4.19</xstream>
 		<pool>2.11.1</pool>
 		<lettuce>6.1.8.RELEASE</lettuce>
-		<jedis>3.8.0</jedis>
+		<jedis>4.2.0</jedis>
 		<multithreadedtc>1.01</multithreadedtc>
 		<netty>4.1.72.Final</netty>
 		<java-module-name>spring.data.redis</java-module-name>

src/main/java/org/springframework/data/redis/connection/convert/SetConverter.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.connection.convert;
 
+import java.util.Collection;
 import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -48,4 +49,8 @@ public Set<T> convert(Set<S> source) {
 		return source.stream().map(itemConverter::convert).collect(Collectors.toCollection(LinkedHashSet::new));
 	}
 
+	public Set<T> convert(Collection<S> source) {
+		return source.stream().map(itemConverter::convert).collect(Collectors.toCollection(LinkedHashSet::new));
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisClientUtils.java

@@ -15,15 +15,13 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
-import redis.clients.jedis.Builder;
-import redis.clients.jedis.Client;
+import redis.clients.jedis.Connection;
 import redis.clients.jedis.Jedis;
-import redis.clients.jedis.Protocol.Command;
-import redis.clients.jedis.Queable;
-import redis.clients.jedis.Response;
-import redis.clients.jedis.util.SafeEncoder;
+import redis.clients.jedis.Protocol;
+import redis.clients.jedis.Transaction;
 
-import java.lang.reflect.Method;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Set;
 import java.util.function.Function;
@@ -43,26 +41,15 @@
 @SuppressWarnings({ "unchecked", "ConstantConditions" })
 class JedisClientUtils {
 
-	private static final Method GET_RESPONSE;
+	private static final Field TRANSACTION;
 	private static final Set<String> KNOWN_COMMANDS;
-	private static final Builder<Object> OBJECT_BUILDER;
 
 	static {
 
-		GET_RESPONSE = ReflectionUtils.findMethod(Queable.class, "getResponse", Builder.class);
-		ReflectionUtils.makeAccessible(GET_RESPONSE);
+		TRANSACTION = ReflectionUtils.findField(Jedis.class, "transaction");
+		ReflectionUtils.makeAccessible(TRANSACTION);
 
-		KNOWN_COMMANDS = Arrays.stream(Command.values()).map(Enum::name).collect(Collectors.toSet());
-
-		OBJECT_BUILDER = new Builder<Object>() {
-			public Object build(Object data) {
-				return data;
-			}
-
-			public String toString() {
-				return "Object";
-			}
-		};
+		KNOWN_COMMANDS = Arrays.stream(Protocol.Command.values()).map(Enum::name).collect(Collectors.toSet());
 	}
 
 	/**
@@ -72,7 +59,7 @@ public String toString() {
 	 * @param keys must not be {@literal null}, may be empty.
 	 * @param args must not be {@literal null}, may be empty.
 	 * @param jedis must not be {@literal null}.
-	 * @return the response, can be be {@literal null}.
+	 * @return the response, can be {@literal null}.
 	 */
 	static <T> T execute(String command, byte[][] keys, byte[][] args, Supplier<Jedis> jedis) {
 		return execute(command, keys, args, jedis, it -> (T) it.getOne());
@@ -86,42 +73,42 @@ static <T> T execute(String command, byte[][] keys, byte[][] args, Supplier<Jedi
 	 * @param args must not be {@literal null}, may be empty.
 	 * @param jedis must not be {@literal null}.
 	 * @param responseMapper must not be {@literal null}.
-	 * @return the response, can be be {@literal null}.
+	 * @return the response, can be {@literal null}.
 	 * @since 2.1
 	 */
 	static <T> T execute(String command, byte[][] keys, byte[][] args, Supplier<Jedis> jedis,
-			Function<Client, T> responseMapper) {
+			Function<Connection, T> responseMapper) {
 
 		byte[][] commandArgs = getCommandArguments(keys, args);
 
-		Client client = sendCommand(command, commandArgs, jedis.get());
+		Connection Connection = sendCommand(command, commandArgs, jedis.get());
 
-		return responseMapper.apply(client);
+		return responseMapper.apply(Connection);
 	}
 
 	/**
-	 * Send a Redis command and retrieve the {@link Client} for response retrieval.
+	 * Send a Redis command and retrieve the {@link Connection} for response retrieval.
 	 *
 	 * @param command the command.
 	 * @param args must not be {@literal null}, may be empty.
 	 * @param jedis must not be {@literal null}.
-	 * @return the {@link Client} instance used to send the command.
+	 * @return the {@link Connection} instance used to send the command.
 	 */
-	static Client sendCommand(String command, byte[][] args, Jedis jedis) {
+	static Connection sendCommand(String command, byte[][] args, Jedis jedis) {
 
-		Client client = jedis.getClient();
+		Connection Connection = jedis.getConnection();
 
-		sendCommand(client, command, args);
+		sendCommand(Connection, command, args);
 
-		return client;
+		return Connection;
 	}
 
-	private static void sendCommand(Client client, String command, byte[][] args) {
+	private static void sendCommand(Connection Connection, String command, byte[][] args) {
 
 		if (isKnownCommand(command)) {
-			client.sendCommand(Command.valueOf(command.trim().toUpperCase()), args);
+			Connection.sendCommand(Protocol.Command.valueOf(command.trim().toUpperCase()), args);
 		} else {
-			client.sendCommand(() -> SafeEncoder.encode(command.trim().toUpperCase()), args);
+			Connection.sendCommand(() -> command.trim().toUpperCase().getBytes(StandardCharsets.UTF_8), args);
 		}
 	}
 
@@ -148,23 +135,13 @@ private static byte[][] getCommandArguments(byte[][] keys, byte[][] args) {
 	}
 
 	/**
-	 * @param jedis the client instance.
+	 * @param jedis the Connection instance.
 	 * @return {@literal true} if the connection has entered {@literal MULTI} state.
 	 */
 	static boolean isInMulti(Jedis jedis) {
-		return jedis.getClient().isInMulti();
-	}
 
-	/**
-	 * Retrieve the {@link Response} object from a {@link redis.clients.jedis.Transaction} or a
-	 * {@link redis.clients.jedis.Pipeline} for response synchronization.
-	 *
-	 * @param target a {@link redis.clients.jedis.Transaction} or {@link redis.clients.jedis.Pipeline}, must not be
-	 *          {@literal null}.
-	 * @return the {@link Response} wrapper object.
-	 */
-	static Response<Object> getResponse(Object target) {
-		return (Response<Object>) ReflectionUtils.invokeMethod(GET_RESPONSE, target, OBJECT_BUILDER);
-	}
+		Object field = ReflectionUtils.getField(TRANSACTION, jedis);
 
+		return field instanceof Transaction;
+	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -15,13 +15,12 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
-import redis.clients.jedis.BinaryJedis;
-import redis.clients.jedis.Client;
+import redis.clients.jedis.Connection;
+import redis.clients.jedis.ConnectionPool;
 import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisCluster;
-import redis.clients.jedis.JedisClusterConnectionHandler;
-import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.providers.ClusterConnectionProvider;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -33,7 +32,6 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
-import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -119,9 +117,13 @@ public JedisClusterConnection(JedisCluster cluster) {
 				new JedisClusterNodeResourceProvider(cluster, topologyProvider), EXCEPTION_TRANSLATION);
 		disposeClusterCommandExecutorOnClose = true;
 
+
 		try {
-			DirectFieldAccessor dfa = new DirectFieldAccessor(cluster);
-			clusterCommandExecutor.setMaxRedirects((Integer) dfa.getPropertyValue("maxRedirections"));
+
+			DirectFieldAccessor executorDfa = new DirectFieldAccessor(cluster);
+			Object custerCommandExecutor = executorDfa.getPropertyValue("executor");
+			DirectFieldAccessor dfa = new DirectFieldAccessor(custerCommandExecutor);
+			clusterCommandExecutor.setMaxRedirects((Integer) dfa.getPropertyValue("maxRedirects"));
 		} catch (Exception e) {
 			// ignore it and work with the executor default
 		}
@@ -177,11 +179,6 @@ public Object execute(String command, byte[]... args) {
 	@Nullable
 	@Override
 	public <T> T execute(String command, byte[] key, Collection<byte[]> args) {
-		return execute(command, key, args, it -> (T) it.getOne());
-	}
-
-	@Nullable
-	<T> T execute(String command, byte[] key, Collection<byte[]> args, Function<Client, T> responseMapper) {
 
 		Assert.notNull(command, "Command must not be null!");
 		Assert.notNull(key, "Key must not be null!");
@@ -191,8 +188,9 @@ <T> T execute(String command, byte[] key, Collection<byte[]> args, Function<Clie
 
 		RedisClusterNode keyMaster = topologyProvider.getTopology().getKeyServingMasterNode(key);
 
-		return clusterCommandExecutor.executeCommandOnSingleNode((JedisClusterCommandCallback<T>) client -> JedisClientUtils
-				.execute(command, EMPTY_2D_BYTE_ARRAY, commandArgs, () -> client, responseMapper), keyMaster).getValue();
+		return clusterCommandExecutor.executeCommandOnSingleNode((JedisClusterCommandCallback<T>) client -> {
+			return (T) client.sendCommand(() -> JedisConverters.toBytes(command), commandArgs);
+		}, keyMaster).getValue();
 	}
 
 	private static byte[][] getCommandArguments(byte[] key, Collection<byte[]> args) {
@@ -409,18 +407,13 @@ public void select(int dbIndex) {
 
 	@Override
 	public byte[] echo(byte[] message) {
-
-		try {
-			return cluster.echo(message);
-		} catch (Exception ex) {
-			throw convertJedisAccessException(ex);
-		}
+		throw new InvalidDataAccessApiUsageException("Echo not supported in cluster mode.");
 	}
 
 	@Override
 	public String ping() {
 
-		return !clusterCommandExecutor.executeCommandOnAllNodes((JedisClusterCommandCallback<String>) BinaryJedis::ping)
+		return !clusterCommandExecutor.executeCommandOnAllNodes((JedisClusterCommandCallback<String>) Jedis::ping)
 				.resultsAsList().isEmpty() ? "PONG" : null;
 
 	}
@@ -429,7 +422,7 @@ public String ping() {
 	public String ping(RedisClusterNode node) {
 
 		return clusterCommandExecutor
-				.executeCommandOnSingleNode((JedisClusterCommandCallback<String>) BinaryJedis::ping, node).getValue();
+				.executeCommandOnSingleNode((JedisClusterCommandCallback<String>) Jedis::ping, node).getValue();
 	}
 
 	/*
@@ -552,8 +545,10 @@ public void clusterReplicate(RedisClusterNode master, RedisClusterNode replica)
 	@Override
 	public Integer clusterGetSlotForKey(byte[] key) {
 
-		return clusterCommandExecutor.executeCommandOnArbitraryNode((JedisClusterCommandCallback<Integer>) client -> client
-				.clusterKeySlot(JedisConverters.toString(key)).intValue()).getValue();
+		return clusterCommandExecutor
+				.executeCommandOnArbitraryNode(
+						(JedisClusterCommandCallback<Integer>) client -> (int) client.clusterKeySlot(JedisConverters.toString(key)))
+				.getValue();
 	}
 
 	@Override
@@ -709,7 +704,7 @@ static class JedisClusterNodeResourceProvider implements ClusterNodeResourceProv
 
 		private final JedisCluster cluster;
 		private final ClusterTopologyProvider topologyProvider;
-		private final JedisClusterConnectionHandler connectionHandler;
+		private final ClusterConnectionProvider connectionHandler;
 
 		/**
 		 * Creates new {@link JedisClusterNodeResourceProvider}.
@@ -726,7 +721,7 @@ static class JedisClusterNodeResourceProvider implements ClusterNodeResourceProv
 
 				PropertyAccessor accessor = new DirectFieldAccessFallbackBeanWrapper(cluster);
 				this.connectionHandler = accessor.isReadableProperty("connectionHandler")
-						? (JedisClusterConnectionHandler) accessor.getPropertyValue("connectionHandler")
+						? (ClusterConnectionProvider) accessor.getPropertyValue("connectionHandler")
 						: null;
 			} else {
 				this.connectionHandler = null;
@@ -739,31 +734,31 @@ public Jedis getResourceForSpecificNode(RedisClusterNode node) {
 
 			Assert.notNull(node, "Cannot get Pool for 'null' node!");
 
-			JedisPool pool = getResourcePoolForSpecificNode(node);
+			ConnectionPool pool = getResourcePoolForSpecificNode(node);
 			if (pool != null) {
-				return pool.getResource();
+				return new Jedis(pool.getResource());
 			}
 
-			Jedis connection = getConnectionForSpecificNode(node);
+			Connection connection = getConnectionForSpecificNode(node);
 
 			if (connection != null) {
-				return connection;
+				return new Jedis(connection);
 			}
 
 			throw new DataAccessResourceFailureException(String.format("Node %s is unknown to cluster", node));
 		}
 
-		private JedisPool getResourcePoolForSpecificNode(RedisClusterNode node) {
+		private ConnectionPool getResourcePoolForSpecificNode(RedisClusterNode node) {
 
-			Map<String, JedisPool> clusterNodes = cluster.getClusterNodes();
+			Map<String, ConnectionPool> clusterNodes = cluster.getClusterNodes();
 			if (clusterNodes.containsKey(node.asString())) {
 				return clusterNodes.get(node.asString());
 			}
 
 			return null;
 		}
 
-		private Jedis getConnectionForSpecificNode(RedisClusterNode node) {
+		private Connection getConnectionForSpecificNode(RedisClusterNode node) {
 
 			RedisClusterNode member = topologyProvider.getTopology().lookup(node);
 
@@ -773,7 +768,7 @@ private Jedis getConnectionForSpecificNode(RedisClusterNode node) {
 			}
 
 			if (member != null && connectionHandler != null) {
-				return connectionHandler.getConnectionFromNode(new HostAndPort(member.getHost(), member.getPort()));
+				return connectionHandler.getConnection(new HostAndPort(member.getHost(), member.getPort()));
 			}
 
 			return null;
@@ -835,15 +830,15 @@ public ClusterTopology getTopology() {
 
 			Map<String, Exception> errors = new LinkedHashMap<>();
 
-			List<Entry<String, JedisPool>> list = new ArrayList<>(cluster.getClusterNodes().entrySet());
+			List<Entry<String, ConnectionPool>> list = new ArrayList<>(cluster.getClusterNodes().entrySet());
 			Collections.shuffle(list);
 
-			for (Entry<String, JedisPool> entry : list) {
+			for (Entry<String, ConnectionPool> entry : list) {
 
-				try (Jedis jedis = entry.getValue().getResource()) {
+				try (Connection connection = entry.getValue().getResource()) {
 
 					time = System.currentTimeMillis();
-					Set<RedisClusterNode> nodes = Converters.toSetOfRedisClusterNodes(jedis.clusterNodes());
+					Set<RedisClusterNode> nodes = Converters.toSetOfRedisClusterNodes(new Jedis(connection).clusterNodes());
 
 					synchronized (lock) {
 						cached = new ClusterTopology(nodes);