Improve safety of MultiNodePipelineBase under multi-thread (#3330)

* Make sure MultiNodePipelineBase is safe under multi-thread * Edit log message --------- Co-authored-by: Stiller <[email protected]> Co-authored-by: M Sazzadul Hoque <[email protected]>
redis · Mar 22, 2023 · 4a8b9e7 · 4a8b9e7
1 parent 281003b
commit 4a8b9e7
Showing 1 changed file with 11 additions and 5 deletions.
diff --git a/src/main/java/redis/clients/jedis/MultiNodePipelineBase.java b/src/main/java/redis/clients/jedis/MultiNodePipelineBase.java
@@ -73,10 +73,16 @@ protected final <T> Response<T> appendCommand(CommandObject<T> commandObject) {
       queue = pipelinedResponses.get(nodeKey);
       connection = connections.get(nodeKey);
     } else {
-      queue = new LinkedList<>();
-      connection = getConnection(nodeKey);
-      pipelinedResponses.put(nodeKey, queue);
-      connections.put(nodeKey, connection);
+      pipelinedResponses.putIfAbsent(nodeKey, new LinkedList<>());
+      queue = pipelinedResponses.get(nodeKey);
+
+      Connection newOne = getConnection(nodeKey);
+      connections.putIfAbsent(nodeKey, newOne);
+      connection = connections.get(nodeKey);
+      if (connection != newOne) {
+        log.debug("Duplicate connection to {}, closing it.", nodeKey);
+        IOUtils.closeQuietly(newOne);
+      }
     }
 
     connection.sendCommand(commandObject.getArguments());
@@ -1005,7 +1011,7 @@ public Response<Double> zscore(String key, String member) {
   }
 
   @Override
-  public Response<List<Double>> zmscore(String key, String... members) {    
+  public Response<List<Double>> zmscore(String key, String... members) {
     return appendCommand(commandObjects.zmscore(key, members));
   }