Align Search APIs and behaviour with Redis 8.0+

- Introduce executeKeylessCommand
- Implement round-robin execution in ClusterCommandExecutor
- Update search commands according to the latest request/response policies
- Remove JedisBroadcastAndRoundRobinConfig
- Introduce AggregateIterator for proper Cluster support
- Deprecate APIs that rely on legacy RediSeach behaviour

Author: uglide

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

@@ -48,8 +48,7 @@ protected RedisProtocol getProtocol() {
   }
 
   protected volatile CommandKeyArgumentPreProcessor keyPreProcessor = null;
-  private JedisBroadcastAndRoundRobinConfig broadcastAndRoundRobinConfig = null;
-  private Lock mapperLock = new ReentrantLock(true);    
+  private Lock mapperLock = new ReentrantLock(true);
   private volatile JsonObjectMapper jsonObjectMapper;
   private final AtomicInteger searchDialect = new AtomicInteger(SearchProtocol.DEFAULT_DIALECT);
 
@@ -58,10 +57,6 @@ public void setKeyArgumentPreProcessor(CommandKeyArgumentPreProcessor keyPreProc
     this.keyPreProcessor = keyPreProcessor;
   }
 
-  void setBroadcastAndRoundRobinConfig(JedisBroadcastAndRoundRobinConfig config) {
-    this.broadcastAndRoundRobinConfig = config;
-  }
-
   protected CommandArguments commandArguments(ProtocolCommand command) {
     CommandArguments comArgs = new CommandArguments(command);
     if (keyPreProcessor != null) comArgs.setKeyArgumentPreProcessor(keyPreProcessor);
@@ -2977,7 +2972,7 @@ public final CommandObject<List<Object>> xreadGroup(byte[] groupName, byte[] con
   }
 
   public final CommandObject<List<Map.Entry<byte[], List<StreamEntryBinary>>>> xreadGroupBinary(
-      byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams, 
+      byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams,
       Map.Entry<byte[], StreamEntryID>... streams) {
     CommandArguments args = commandArguments(XREADGROUP)
         .add(GROUP).add(groupName).add(consumer)
@@ -2992,7 +2987,7 @@ public final CommandObject<List<Map.Entry<byte[], List<StreamEntryBinary>>>> xre
   }
 
   public final CommandObject<Map<byte[], List<StreamEntryBinary>>> xreadGroupBinaryAsMap(
-      byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams, 
+      byte[] groupName, byte[] consumer, XReadGroupParams xReadGroupParams,
       Map.Entry<byte[], StreamEntryID>... streams) {
     CommandArguments args = commandArguments(XREADGROUP)
         .add(GROUP).add(groupName).add(consumer)
@@ -3481,18 +3476,16 @@ public final CommandObject<Long> hsetObject(String key, Map<String, Object> hash
     return new CommandObject<>(addFlatMapArgs(commandArguments(HSET).key(key), hash), BuilderFactory.LONG);
   }
 
-  private boolean isRoundRobinSearchCommand() {
-    if (broadcastAndRoundRobinConfig == null) {
-      return true;
-    } else if (broadcastAndRoundRobinConfig.getRediSearchModeInCluster() == JedisBroadcastAndRoundRobinConfig.RediSearchMode.LIGHT) {
-      return false;
-    }
-    return true;
+  private boolean isRoundRobinSearchCommand(SearchCommand sc) {
+
+    return !(sc.equals(SearchCommand.SUGGET) || sc.equals(SearchCommand.SUGADD) || sc.equals(
+        SearchCommand.SUGLEN) || sc.equals(SearchCommand.SUGDEL) || sc.equals(
+        SearchCommand.CURSOR));
   }
 
   private CommandArguments checkAndRoundRobinSearchCommand(SearchCommand sc, String idx) {
     CommandArguments ca = commandArguments(sc);
-    if (isRoundRobinSearchCommand()) {
+    if (isRoundRobinSearchCommand(sc)) {
       ca.add(idx);
     } else {
       ca.key(idx);
@@ -3502,16 +3495,22 @@ private CommandArguments checkAndRoundRobinSearchCommand(SearchCommand sc, Strin
 
   private CommandArguments checkAndRoundRobinSearchCommand(SearchCommand sc, String idx1, String idx2) {
     CommandArguments ca = commandArguments(sc);
-    if (isRoundRobinSearchCommand()) {
+    if (isRoundRobinSearchCommand(sc)) {
       ca.add(idx1).add(idx2);
     } else {
       ca.key(idx1).key(idx2);
     }
     return ca;
   }
 
-  private CommandArguments checkAndRoundRobinSearchCommand(CommandArguments commandArguments, byte[] indexName) {
-    return isRoundRobinSearchCommand() ? commandArguments.add(indexName) : commandArguments.key(indexName);
+  private CommandArguments checkAndRoundRobinSearchCommand(SearchCommand sc, byte[] indexName) {
+    CommandArguments ca = commandArguments(sc);
+    if (isRoundRobinSearchCommand(sc)) {
+      ca.add(indexName);
+    } else {
+      ca.key(indexName);
+    }
+    return ca;
   }
 
   private <T> CommandObject<T> directSearchCommand(CommandObject<T> object, String indexName) {
@@ -3592,7 +3591,7 @@ public final CommandObject<SearchResult> ftSearch(byte[] indexName, Query query)
     if (protocol == RedisProtocol.RESP3) {
       throw new UnsupportedOperationException("binary ft.search is not implemented with resp3.");
     }
-    return new CommandObject<>(checkAndRoundRobinSearchCommand(commandArguments(SearchCommand.SEARCH), indexName)
+    return new CommandObject<>(checkAndRoundRobinSearchCommand(SearchCommand.SEARCH, indexName)
         .addParams(query.dialectOptional(searchDialect.get())), getSearchResultBuilder(null,
         () -> new SearchResultBuilder(!query.getNoContent(), query.getWithScores(), false)));
   }