Introduce request & response policy support

.github/workflows/integration.yml

@@ -11,6 +11,7 @@ on:
     branches:
       - master
       - '[0-9].*'
+      - 'topic/**'
   pull_request:
     branches:
       - master

.github/workflows/test-on-docker.yml

@@ -13,6 +13,7 @@ on:
       - master
       - '[0-9].*'
       - 'feature/**'
+      - 'topic/**'
   pull_request:
     branches:
       - master

pom.xml

@@ -568,6 +568,8 @@
 						<include>**/resps/LibraryInfoTest.java</include>
 						<include>**/*Matchers.java</include>
 						<include>**/*TestUtil.java</include>
+						<include>**/executors/aggregators/*.java</include>
+						<include>**/*MapMatcher.java</include>
 					</includes>
 				</configuration>
 				<executions>

src/main/java/redis/clients/jedis/ClusterPipeline.java

@@ -3,9 +3,58 @@
 import java.time.Duration;
 import java.util.Set;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import redis.clients.jedis.exceptions.JedisClusterOperationException;
 import redis.clients.jedis.providers.ClusterConnectionProvider;
 import redis.clients.jedis.util.IOUtils;
 
+/**
+ * Pipeline implementation for Redis Cluster mode.
+ * <p>
+ * ClusterPipeline allows batching multiple commands for efficient execution in a Redis Cluster
+ * environment. Commands are automatically routed to the appropriate cluster nodes based on
+ * key hash slots.
+ * </p>
+ * <p>
+ * <strong>Important Limitations:</strong>
+ * </p>
+ * <ul>
+ * <li><strong>Single-node commands only:</strong> Only commands that can be routed to a single
+ * node are supported. Commands requiring execution on multiple nodes (ALL_SHARDS, MULTI_SHARD,
+ * ALL_NODES, or SPECIAL request policies) will throw {@link UnsupportedOperationException}.</li>
+ * <li><strong>Examples of unsupported commands:</strong>
+ *   <ul>
+ *     <li>{@code KEYS} - requires execution on all master shards</li>
+ *     <li>{@code MGET} with keys in different slots - requires execution on multiple shards</li>
+ *     <li>{@code SCRIPT LOAD} - requires execution on all nodes</li>
+ *   </ul>
+ * </li>
+ * <li>For multi-node commands, use the non-pipelined mode
+ * of {@link RedisClusterClient} instead.</li>
+ * </ul>
+ * <p>
+ * <strong> Usage Pattern:</strong>
+ * </p>
+ * <pre>{@code
+ * try (RedisCluster cluster = new RedisCluster(nodes, config)) {
+ *   // For single-node commands, use pipelined mode
+ *   try (ClusterPipeline pipeline = cluster.pipelined()) {
+ *     Response<String> r1 = pipeline.set("key1", "value1");
+ *     Response<String> r2 = pipeline.get("key1");
+ *     pipeline.sync();
+ *
+ *     System.out.println(r1.get()); // "OK"
+ *     System.out.println(r2.get()); // "value1"
+ *   }
+ *
+ *   // For multi-node commands, use non-pipelined mode
+ *   Set<String> allKeys = cluster.keys("*"); // Executes on all master shards
+ *   List<String> values = cluster.mget("key1", "key2", "key3"); // Cross-slot keys
+ * }
+ * }</pre>
+ *
+ * @see MultiNodePipelineBase
+ * @see redis.clients.jedis.RedisClusterClient
+ */
 public class ClusterPipeline extends MultiNodePipelineBase {
 
   private final ClusterConnectionProvider provider;
@@ -40,6 +89,12 @@ public ClusterPipeline(ClusterConnectionProvider provider, ClusterCommandObjects
     this.provider = provider;
   }
 
+  ClusterPipeline(ClusterConnectionProvider provider, ClusterCommandObjects commandObjects,
+      CommandFlagsRegistry commandFlagsRegistry) {
+    super(commandObjects, commandFlagsRegistry);
+    this.provider = provider;
+  }
+
   private static ClusterCommandObjects createClusterCommandObjects(RedisProtocol protocol) {
     ClusterCommandObjects cco = new ClusterCommandObjects();
     if (protocol == RedisProtocol.RESP3) cco.setProtocol(protocol);
@@ -57,7 +112,17 @@ public void close() {
 
   @Override
   protected HostAndPort getNodeKey(CommandArguments args) {
-    return provider.getNode(((ClusterCommandArguments) args).getCommandHashSlot());
+    Set<Integer> slots = args.getKeyHashSlots();
+
+    if (slots.size() > 1) {
+      throw new JedisClusterOperationException("Cannot get NodeKey for command with multiple hash slots");
+    }
+
+    if (slots.isEmpty()) {
+      return null; // Let getConnection(null) handle it by using a random node
+    }
+
+    return provider.getNode(slots.iterator().next());
   }
 
   @Override

src/main/java/redis/clients/jedis/CommandArguments.java

@@ -1,11 +1,6 @@
 package redis.clients.jedis;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
+import java.util.*;
 
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.annots.Internal;
@@ -14,13 +9,30 @@
 import redis.clients.jedis.commands.ProtocolCommand;
 import redis.clients.jedis.params.IParams;
 import redis.clients.jedis.search.RediSearchUtil;
+import redis.clients.jedis.util.JedisClusterCRC16;
 
 public class CommandArguments implements Iterable<Rawable> {
 
+  /**
+   * Default initial capacity for the keys list. Most Redis commands have 1-3 keys,
+   * so a small initial capacity avoids reallocations for common cases.
+   */
+  private static final int DEFAULT_KEYS_CAPACITY = 4;
+
   private CommandKeyArgumentPreProcessor keyPreProc = null;
   private final ArrayList<Rawable> args;
 
-  private List<Object> keys;
+  /**
+   * Pre-allocated list for storing keys. Using ArrayList directly avoids the
+   * memory reallocation overhead of transitioning from emptyList -> singletonList -> ArrayList.
+   */
+  private final ArrayList<Object> keys;
+
+  /**
+   * Cached hash slots computed from keys. Null indicates the cache is invalid
+   * and needs to be recomputed. The cache is invalidated when keys are added.
+   */
+  private Set<Integer> cachedHashSlots;
 
   private boolean blocking;
 
@@ -32,7 +44,8 @@ public CommandArguments(ProtocolCommand command) {
     args = new ArrayList<>();
     args.add(command);
 
-    keys = Collections.emptyList();
+    keys = new ArrayList<>(DEFAULT_KEYS_CAPACITY);
+    cachedHashSlots = null;
   }
 
   public ProtocolCommand getCommand() {
@@ -120,36 +133,36 @@ public CommandArguments key(Object key) {
 
     if (key instanceof Rawable) {
       Rawable raw = (Rawable) key;
-      processKey(raw.getRaw());
       args.add(raw);
+      // Extract raw bytes for hash slot computation to avoid ClassCastException in getKeyHashSlots()
+      addHashSlotKey(raw.getRaw());
     } else if (key instanceof byte[]) {
       byte[] raw = (byte[]) key;
-      processKey(raw);
       args.add(RawableFactory.from(raw));
+      addHashSlotKey(raw);
     } else if (key instanceof String) {
       String raw = (String) key;
-      processKey(raw);
       args.add(RawableFactory.from(raw));
+      addHashSlotKey(raw);
     } else {
       throw new IllegalArgumentException("\"" + key.toString() + "\" is not a valid argument.");
     }
 
-    addKeyInKeys(key);
+    return this;
+  }
 
+  final CommandArguments addHashSlotKey(String key) {
+    keys.add(key);
+    // Invalidate cached hash slots since keys have changed
+    cachedHashSlots = null;
     return this;
   }
 
-  private void addKeyInKeys(Object key) {
-    if (keys.isEmpty()) {
-      keys = Collections.singletonList(key);
-    } else if (keys.size() == 1) {
-      List oldKeys = keys;
-      keys = new ArrayList();
-      keys.addAll(oldKeys);
-      keys.add(key);
-    } else {
-      keys.add(key);
-    }
+  final CommandArguments addHashSlotKey(byte[] key) {
+    keys.add(key);
+    // Invalidate cached hash slots since keys have changed
+    cachedHashSlots = null;
+    return this;
   }
 
   public final CommandArguments keys(Object... keys) {
@@ -167,26 +180,16 @@ public final CommandArguments addParams(IParams params) {
     return this;
   }
 
-  protected CommandArguments processKey(byte[] key) {
-    // do nothing
-    return this;
-  }
-
-  protected final CommandArguments processKeys(byte[]... keys) {
+  protected final CommandArguments addHashSlotKeys(byte[]... keys) {
     for (byte[] key : keys) {
-      processKey(key);
+      addHashSlotKey(key);
     }
     return this;
   }
 
-  protected CommandArguments processKey(String key) {
-    // do nothing
-    return this;
-  }
-
-  protected final CommandArguments processKeys(String... keys) {
+  protected final CommandArguments addHashSlotKeys(String... keys) {
     for (String key : keys) {
-      processKey(key);
+      addHashSlotKey(key);
     }
     return this;
   }
@@ -210,9 +213,57 @@ public Iterator<Rawable> iterator() {
     return args.iterator();
   }
 
+  /**
+   * Returns the keys used in this command.
+   * <p>
+   * <b>Internal API:</b> This method is internal and should not be used by external code.
+   * It is exposed for internal use by caching ({@link redis.clients.jedis.csc.CacheKey#getRedisKeys()})
+   * and cluster operations.
+   * <p>
+   * <b>Supported types:</b> Keys are stored as either {@link String} or {@code byte[]} depending on
+   * how they were added via {@link #key(Object)} or {@link #addHashSlotKey(String)}/{@link #addHashSlotKey(byte[])}.
+   * Only {@link String} and {@code byte[]} are guaranteed to be supported by downstream consumers.
+   * <p>
+   * <b>Type safety:</b> Consumers must handle both {@link String} and {@code byte[]} types.
+   * Passing other types may cause {@link IllegalArgumentException} when used with caching
+   * (see {@link redis.clients.jedis.csc.AbstractCache#makeKeyForRedisKeysToCacheKeys(Object)})
+   * or cluster operations.
+   * <p>
+   * The returned list is unmodifiable to prevent external modification of the internal key tracking.
+   *
+   * @return unmodifiable list of keys ({@link String} or {@code byte[]})
+   */
   @Internal
   public List<Object> getKeys() {
-    return keys;
+    return Collections.unmodifiableList(keys);
+  }
+
+  @Internal
+  public Set<Integer> getKeyHashSlots() {
+    // Return cached slots if available (cache is invalidated when keys are added)
+    if (cachedHashSlots != null) {
+      return cachedHashSlots;
+    }
+
+    // Compute hash slots and cache the result
+    Set<Integer> slots = new HashSet<>();
+    for (Object key : keys) {
+      if (key instanceof byte[]) {
+        slots.add(JedisClusterCRC16.getSlot((byte[]) key));
+      } else {
+        slots.add(JedisClusterCRC16.getSlot((String) key));
+      }
+    }
+    // Cache as unmodifiable set to prevent external modification
+    cachedHashSlots = Collections.unmodifiableSet(slots);
+    return cachedHashSlots;
+  }
+
+  /**
+   * @return true if this command has no keys, false otherwise
+   */
+  public boolean isKeyless() {
+    return keys.isEmpty();
   }
 
   public boolean isBlocking() {