Fixed - RLiveObjectService.persist() method with varargs hangs in cluster mode. #5449

redisson/src/main/java/org/redisson/liveobject/core/AccessorInterceptor.java

@@ -38,6 +38,7 @@ import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.util.*;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
 import java.util.regex.Pattern;
 
 /**
@@ -204,15 +205,23 @@ public class AccessorInterceptor {
         } else {
             if (ClassUtils.isAnnotationPresent(field.getType(), REntity.class)
                     || commandExecutor.getConnectionManager().isClusterMode()) {
-                Object value = liveMap.remove(field.getName());
-                if (value != null) {
-                    RMultimapAsync<Object, Object> map = new RedissonSetMultimap<>(namingScheme.getCodec(), ce, indexName);
-                    Object k = value;
-                    if (ClassUtils.isAnnotationPresent(field.getType(), REntity.class)) {
-                        k = ((RLiveObject) value).getLiveObjectId();
-                    }
-                    map.removeAsync(k, ((RLiveObject) me).getLiveObjectId());
+                CompletableFuture<Object> f;
+                if (commandExecutor instanceof CommandBatchService) {
+                    f = liveMap.removeAsync(field.getName()).toCompletableFuture();
+                } else {
+                    Object value = liveMap.remove(field.getName());
+                    f = CompletableFuture.completedFuture(value);
                 }
+                f.thenAccept(value -> {
+                    if (value != null) {
+                        RMultimapAsync<Object, Object> map = new RedissonSetMultimap<>(namingScheme.getCodec(), ce, indexName);
+                        Object k = value;
+                        if (ClassUtils.isAnnotationPresent(field.getType(), REntity.class)) {
+                            k = ((RLiveObject) value).getLiveObjectId();
+                        }
+                        map.removeAsync(k, ((RLiveObject) me).getLiveObjectId());
+                    }
+                });
             } else {
                 removeAsync(ce, indexName, ((RedissonObject) liveMap).getRawName(),
                         namingScheme.getCodec(), ((RLiveObject) me).getLiveObjectId(), field.getName());

redisson/src/test/java/org/redisson/RedissonLiveObjectServiceTest.java

@@ -6,12 +6,10 @@ import org.junit.jupiter.api.Timeout;
 import org.redisson.api.*;
 import org.redisson.api.annotation.*;
 import org.redisson.api.condition.Conditions;
-import org.redisson.config.Config;
 import org.redisson.liveobject.resolver.DefaultNamingScheme;
 import org.redisson.liveobject.resolver.LongGenerator;
 import org.redisson.liveobject.resolver.UUIDGenerator;
 
-import java.io.IOException;
 import java.io.Serializable;
 import java.time.Duration;
 import java.util.*;
@@ -23,6 +21,7 @@ import static org.junit.jupiter.api.Assertions.*;
 /**
  *
  * @author Rui Gu (https://github.com/jackygurui)
+ * @author Nikita Koksharov
  */
 public class RedissonLiveObjectServiceTest extends RedisDockerTest {
 
@@ -730,6 +729,21 @@ public class RedissonLiveObjectServiceTest extends RedisDockerTest {
         assertThat(ids3).isEqualTo(2);
     }
 
+    @Test
+    public void testPersistInCluster() {
+        testInCluster(redisson -> {
+            RLiveObjectService liveObjectService = redisson.getLiveObjectService();
+            TestIndexed item1 = new TestIndexed("1");
+            item1.setName1("name1");
+            item1.setName2("name2");
+            item1.setNum1(123);
+
+            TestIndexed item2 = new TestIndexed("2");
+
+            liveObjectService.persist(item1, item2);
+        });
+    }
+
     @Test
     public void testIndexUpdateCluster() {
         testInCluster(redisson -> {