IGNITE-11754 Fixed memory leak in TxFinishSync - Fixes apache#6462.

Signed-off-by: Alexey Goncharuk <alexey.goncharuk@gmail.com>

(cherry-picked from commit #3a392609)

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxFinishSync.java

@@ -29,7 +29,6 @@
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.lang.IgniteFuture;
 import org.jetbrains.annotations.Nullable;
 
@@ -66,9 +65,16 @@ public void onFinishSend(UUID nodeId, long threadId) {
         ThreadFinishSync threadSync = threadMap.get(threadId);
 
         if (threadSync == null)
-            threadSync = F.addIfAbsent(threadMap, threadId, new ThreadFinishSync(threadId));
+            threadMap.put(threadId, threadSync = new ThreadFinishSync(threadId));
 
-        threadSync.onSend(nodeId);
+        synchronized (threadSync) {
+            //thread has to create new ThreadFinishSync
+            //if other thread executing onAckReceived method removed previous threadSync object
+            if (threadMap.get(threadId) == null)
+                threadMap.put(threadId, threadSync = new ThreadFinishSync(threadId));
+
+            threadSync.onSend(nodeId);
+        }
     }
 
     /**
@@ -104,8 +110,14 @@ public void onDisconnected(IgniteFuture<?> reconnectFut) {
     public void onAckReceived(UUID nodeId, long threadId) {
         ThreadFinishSync threadSync = threadMap.get(threadId);
 
-        if (threadSync != null)
+        if (threadSync != null) {
             threadSync.onReceive(nodeId);
+
+            synchronized (threadSync) {
+                if (threadSync.isEmpty())
+                    threadMap.remove(threadId);
+            }
+        }
     }
 
     /**
@@ -114,8 +126,14 @@ public void onAckReceived(UUID nodeId, long threadId) {
      * @param nodeId Left node ID.
      */
     public void onNodeLeft(UUID nodeId) {
-        for (ThreadFinishSync threadSync : threadMap.values())
+        for (ThreadFinishSync threadSync : threadMap.values()) {
             threadSync.onNodeLeft(nodeId);
+
+            synchronized (threadSync) {
+                if (threadSync.isEmpty())
+                    threadMap.remove(threadSync);
+            }
+        }
     }
 
     /**
@@ -193,7 +211,7 @@ public void onDisconnected(IgniteFuture<?> reconnectFut) {
          * @param nodeId Node ID response received from.
          */
         public void onReceive(UUID nodeId) {
-            TxFinishSync sync = nodeMap.get(nodeId);
+            TxFinishSync sync = nodeMap.remove(nodeId);
 
             if (sync != null)
                 sync.onReceive();
@@ -208,6 +226,13 @@ public void onNodeLeft(UUID nodeId) {
             if (sync != null)
                 sync.onNodeLeft();
         }
+
+        /**
+         *
+         */
+        private boolean isEmpty() {
+            return nodeMap.isEmpty();
+        }
     }
 
     /**