Reactive Cluster `MGET` is not running in parallel

[koisyu]

Bug Report

RedisAdvancedClusterReactiveCommandsImpl.mget is not running in parallel

Current Behavior

If you run the code below, it will take a long time.

RedisClusterClient client = RedisClusterClient.create("redis://localhost");
RedisAdvancedClusterReactiveCommands<String, String> reactiveCommands = client.connect().reactive();
List<String> keys = IntStream.rangeClosed(1, 100).boxed().map(index -> "key-" + index).collect(Collectors.toList());
reactiveCommands.mget(keys.toArray(new String[]{})).collectList().block();

If I run the above code and check the Tcpdump, I can see that the requests and responses are alternating in order, like this.
That is, they are executed sequentially.

Expected behavior/code

I expect MGET commands to be executed in parallel.

Environment

• Lettuce version(s): 6.1.9.RELEASE
• Redis version: 4.0.14

Possible Solution

If the Flux#concat code is replaced with Flux#mergeSequential, the MGET commands are executed in parallel.

• Flux#concat
  
• Flux#mergeSequential
  

Additional context

[koisyu]

I was worried that my explanation wasn't clear enough, so I added an example of the difference in repeated runs.
For example, if you run the code below, you can clearly see the difference.

RedisClusterClient client = RedisClusterClient.create("redis://localhost");
StatefulRedisClusterConnection<String, String> connect = client.connect();
RedisAdvancedClusterReactiveCommands<String, String> reactiveCommands = connect.reactive();
String[] keys = IntStream.rangeClosed(1, 500).boxed().map(index -> "key-" + index).toArray(String[]::new);
reactiveCommands.mget(keys).collectList().block();

long startTimeInMillis = System.currentTimeMillis();
for (int i = 0; i < 100; i++) {
    reactiveCommands.mget(keys).collectList().block();
}
long endTimeInMillis = System.currentTimeMillis();
System.out.println(String.format("Total time in millis : %s", (endTimeInMillis - startTimeInMillis)));

Here are the execution results of the code in both cases. The execution times vary slightly depending on the conditions, but the difference between the two is always significant.

• Flux#concat
  • Total time in millis : 8289

• Flux#mergeSequential
  • Total time in millis : 563

[wenhaozhao]

@mp911de Can fix the same bug to version 6.1.x ?

[tishun]

@mp911de Can fix the same bug to version 6.1.x ?

Hey @wenhaozhao , is this still relevant or can we close this issue?

[wenhaozhao]

@mp911de Can fix the same bug to version 6.1.x ?

Hey @wenhaozhao , is this still relevant or can we close this issue?

sure, please