Repository: hazelcast/hazelcast (language: java)

Issue:
Client and server memory leaks while destroying FencedLocks [HZ-3039]
The local `FencedLock` proxy object is not removed from `LockService#proxies` [map](https://github.com/hazelcast/hazelcast/blob/master/hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/lock/LockService.java#L57) while the `Lock#destroy()` called
 
Steps to reproduce:
```
        HazelcastInstance instance = Hazelcast.newHazelcastInstance(config);
        Hazelcast.newHazelcastInstance(config);
        Hazelcast.newHazelcastInstance(config);
        CPSubsystem cpSubsystem = instance.getCPSubsystem();

        for (int i = 0; i < 100; i++) {
            FencedLock lock = cpSubsystem.getLock("iLock_" + i);
            lock.lock();
            lock.unlock();
            lock.destroy();
        }
```
The `LockService#proxies` map contains 100 FencedLockProxy objects:

<img width="374" alt="image" src="https://github.com/hazelcast/hazelcast/assets/1538227/01d448a5-0b45-4e91-842d-538b7101ebe9">

This behaviour causes memory leaks.

The same behaviour is also on the client side: the `ClientRaftProxyFactory#lockProxies` tracks all created `FencedLockProxy` and not remove from the map.

The same client-side `_lock_proxies` map can be found in the Python client https://github.com/hazelcast/hazelcast-python-client/blob/master/hazelcast/cp.py#L198


Discussion / hints:
From code we only tidy `LockService.proxies` up under the following scenarios:

1. CP system restart, `LockService#onCPSubsystemRestart()`; and
2. When a proxy with the same name already exists but is associated with an unexpected CP group ID

Doesn't look like `proxies` mapping is critical to other semantics.

If the current semantics of not permitting the same object to be reused post-destruction are to be preserved and we don't want the removal from `proxies` to deviate from current semantics, then @arodionov is it possible the safest (read: most conservative) place to perform the removal would be _after_ we add the object name to `ResourceRegistry.destroyedNames`? Looking at the code I think this could be done earlier but not sure if that will have some unknown side effect.

As a sketch:

```
class LockService ... {
    // ...
    @Override
    public boolean destroyRaftObject(CPGroupId groupId, String name) {
        boolean result = super.destroyRaftObject(groupId, name);
        FencedLockProxy proxy = proxies.get(name);
        proxies.remove(name, proxy);
        return result;
    }
}
```

Fix patch:
diff --git a/hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/lock/LockService.java b/hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/lock/LockService.java
index f7b4f2f29277e..10d56db6361d2 100644
--- a/hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/lock/LockService.java
+++ b/hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/lock/LockService.java
@@ -244,4 +244,11 @@ public void provideDynamicMetrics(MetricDescriptor descriptor, MetricsCollection
         }
     }
 
+    @Override
+    public boolean destroyRaftObject(CPGroupId groupId, String name) {
+        boolean result = super.destroyRaftObject(groupId, name);
+        String proxyName = withoutDefaultGroupName(name + "@" + groupId.getName());
+        proxies.remove(proxyName);
+        return result;
+    }
 }


Test patch:
diff --git a/hazelcast/src/test/java/com/hazelcast/cp/internal/datastructures/lock/FencedLockProxyRemovalTest.java b/hazelcast/src/test/java/com/hazelcast/cp/internal/datastructures/lock/FencedLockProxyRemovalTest.java
new file mode 100644
index 0000000000000..b797220fd55ff
--- /dev/null
+++ b/hazelcast/src/test/java/com/hazelcast/cp/internal/datastructures/lock/FencedLockProxyRemovalTest.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2008-2023, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hazelcast.cp.internal.datastructures.lock;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.cp.internal.HazelcastRaftTestSupport;
+import com.hazelcast.cp.internal.datastructures.lock.proxy.FencedLockProxy;
+import com.hazelcast.cp.lock.FencedLock;
+import com.hazelcast.test.HazelcastParallelClassRunner;
+import com.hazelcast.test.annotation.ParallelJVMTest;
+import com.hazelcast.test.annotation.QuickTest;
+import com.hazelcast.test.starter.ReflectionUtils;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.concurrent.ConcurrentMap;
+
+import static com.hazelcast.test.Accessors.getNodeEngineImpl;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(HazelcastParallelClassRunner.class)
+@Category({QuickTest.class, ParallelJVMTest.class})
+public class FencedLockProxyRemovalTest extends HazelcastRaftTestSupport {
+    protected HazelcastInstance[] instances;
+
+    @Before
+    public void setup() {
+        instances = newInstances(3);
+    }
+
+    @Test
+    public void testDestroy_AllMembersRemoveLockProxyDefaultCpGroup() {
+        allMembersRemoveLockProxy(randomName());
+    }
+
+    @Test
+    public void testDestroy_AllMembersRemoveLockProxyCustomCpGroup() {
+        allMembersRemoveLockProxy(randomName() + "@mygroup");
+    }
+
+    private void allMembersRemoveLockProxy(String lockName) {
+        FencedLock myLockMember0 = instances[0].getCPSubsystem().getLock(lockName);
+        FencedLock myLockMember1 = instances[1].getCPSubsystem().getLock(lockName);
+        FencedLock myLockMember2 = instances[2].getCPSubsystem().getLock(lockName);
+
+        myLockMember0.lock();
+        try {
+        } finally {
+            myLockMember0.unlock();
+        }
+        myLockMember0.destroy();
+
+        for (HazelcastInstance instance : instances) {
+            assertTrueEventually(() -> {
+                LockService service = getNodeEngineImpl(instance).getService(LockService.SERVICE_NAME);
+                ConcurrentMap<String, FencedLockProxy> proxies = ReflectionUtils.getFieldValueReflectively(service, "proxies");
+                assertTrue(proxies.isEmpty());
+            });
+        }
+    }
+}