Merge pull request #13755 from sancar/fix/letEndpointReferenceGoWhenC…

…lientDisconnected/maint3.x

Dont keep the endpoint reference to clean client resources

hazelcast/src/main/java/com/hazelcast/client/impl/ClientEngineImpl.java

@@ -460,13 +460,15 @@ public void connectionRemoved(Connection connection) {
             }
 
             String localMemberUuid = node.getThisUuid();
-            String ownerUuid = ownershipMappings.get(endpoint.getUuid());
+            final String clientUuid = endpoint.getUuid();
+            String ownerUuid = ownershipMappings.get(clientUuid);
             if (localMemberUuid.equals(ownerUuid)) {
+                final long authenticationCorrelationId = endpoint.getAuthenticationCorrelationId();
                 try {
                     nodeEngine.getExecutionService().schedule(new Runnable() {
                         @Override
                         public void run() {
-                            callDisconnectionOperation(endpoint);
+                            callDisconnectionOperation(clientUuid, authenticationCorrelationId);
                         }
                     }, endpointRemoveDelaySeconds, TimeUnit.SECONDS);
                 } catch (RejectedExecutionException e) {
@@ -477,19 +479,18 @@ public void run() {
             }
         }
 
-        private void callDisconnectionOperation(ClientEndpointImpl endpoint) {
+        private void callDisconnectionOperation(String clientUuid, long authenticationCorrelationId) {
             Collection<Member> memberList = nodeEngine.getClusterService().getMembers();
             OperationService operationService = nodeEngine.getOperationService();
             String memberUuid = getLocalMember().getUuid();
-            String clientUuid = endpoint.getUuid();
 
             String ownerMember = ownershipMappings.get(clientUuid);
             if (!memberUuid.equals(ownerMember)) {
                 // do nothing if the owner already changed (double checked locking)
                 return;
             }
 
-            if (lastAuthenticationCorrelationIds.get(clientUuid).get() > endpoint.getAuthenticationCorrelationId()) {
+            if (lastAuthenticationCorrelationIds.get(clientUuid).get() > authenticationCorrelationId) {
                 //a new authentication already made for that client. This check is needed to detect
                 // "a disconnected client is reconnected back to same node"
                 return;