Fixed - semaphore object is not deleted after RLocalCachedMap.clearLocalCache method invocation. #2273

redisson/src/main/java/org/redisson/cache/LocalCacheListener.java

@@ -277,18 +277,24 @@ public abstract class LocalCacheListener {
         }
 
         byte[] id = commandExecutor.getServiceManager().generateIdArray();
-        RFuture<Long> future = publishAsync(id);
-        CompletionStage<Void> f = future.thenCompose(res -> {
-            if (res.intValue() == 0) {
-                return CompletableFuture.completedFuture(null);
-            }
+        RSemaphore semaphore = getClearSemaphore(id);
+        CompletionStage<Void> f = semaphore.trySetPermitsAsync(0)
+                .thenCompose(r -> semaphore.expireAsync(Duration.ofSeconds(60)))
+                .thenCompose(r -> publishAsync(id))
+                .thenCompose(res -> {
+                    if (res == 0) {
+                        return semaphore.deleteAsync()
+                                        .thenApply(r -> null);
+                    }
 
-            RSemaphore semaphore = getClearSemaphore(id);
-            return semaphore.tryAcquireAsync(res.intValue() - 1, 50, TimeUnit.SECONDS)
-                    .thenCompose(r -> {
-                        return semaphore.deleteAsync().thenApply(re -> null);
-                    });
-        });
+                    System.out.println("res " + res);
+                    return semaphore.tryAcquireAsync(res.intValue() - 1, 40, TimeUnit.SECONDS)
+                                    .thenCompose(r -> {
+                                        System.out.println("aca " + r);
+                                        return semaphore.deleteAsync()
+                                                .thenApply(re -> null);
+                                    });
+                });
         return new CompletableFutureWrapper<>(f);
     }
 
@@ -382,9 +388,7 @@ public abstract class LocalCacheListener {
 
     private RSemaphore getClearSemaphore(byte[] requestId) {
         String id = ByteBufUtil.hexDump(requestId);
-        RSemaphore semaphore = new RedissonSemaphore(commandExecutor, name + ":clear:" + id);
-        semaphore.expireAsync(Duration.ofSeconds(60));
-        return semaphore;
+        return new RedissonSemaphore(commandExecutor, name + ":clear:" + id);
     }
 
     public <K, V> int addListener(LocalCacheInvalidateListener<K, V> listener) {

redisson/src/test/java/org/redisson/RedissonLocalCachedMapTest.java

@@ -557,11 +557,14 @@ public class RedissonLocalCachedMapTest extends BaseMapTest {
                 .evictionPolicy(EvictionPolicy.LFU)
                 .cacheSize(5)
                 .syncStrategy(SyncStrategy.INVALIDATE);
-        
+
+        Config c = redisson.getConfig();
+        RedissonClient redisson2 = Redisson.create(c);
+
         RLocalCachedMap<String, Integer> map1 = redisson.getLocalCachedMap(options);
         Map<String, Integer> cache1 = map1.getCachedMap();
         
-        RLocalCachedMap<String, Integer> map2 = redisson.getLocalCachedMap(options);
+        RLocalCachedMap<String, Integer> map2 = redisson2.getLocalCachedMap(options);
         Map<String, Integer> cache2 = map2.getCachedMap();
         
         map1.put("1", 1);
@@ -573,13 +576,14 @@ public class RedissonLocalCachedMapTest extends BaseMapTest {
         assertThat(cache1.size()).isEqualTo(2);
         assertThat(cache2.size()).isEqualTo(2);
 
-        
         map1.clearLocalCache();
-        
+
         assertThat(redisson.getKeys().count()).isEqualTo(1);
 
         assertThat(cache1.size()).isZero();
         assertThat(cache2.size()).isZero();
+
+        redisson2.shutdown();
     }