From 23af8bf321470a3274aa8c131e5830d2ef0fa05a Mon Sep 17 00:00:00 2001
From: sancar <sancar@hazelcast.com>
Date: Mon, 3 Sep 2018 12:09:56 +0300
Subject: [PATCH] Fixing heartbeat resume condition

With the fix #13578, the condition of resuming heartbeat is became
broken.

It is fixed so that, heartbeat of a connection becomes healthy when
last read time is lower than heartbeat timeout.
---
 .../client/connection/nio/HeartbeatManager.java      | 12 ++++++------
 .../client/heartbeat/ClientHeartbeatTest.java        | 11 ++++++++---
 2 files changed, 14 insertions(+), 9 deletions(-)

diff --git a/hazelcast-client/src/main/java/com/hazelcast/client/connection/nio/HeartbeatManager.java b/hazelcast-client/src/main/java/com/hazelcast/client/connection/nio/HeartbeatManager.java
index 27a82564b093..be00f5e613ed 100644
--- a/hazelcast-client/src/main/java/com/hazelcast/client/connection/nio/HeartbeatManager.java
+++ b/hazelcast-client/src/main/java/com/hazelcast/client/connection/nio/HeartbeatManager.java
@@ -97,6 +97,12 @@ private void checkConnection(long now, final ClientConnection connection) {
                 connection.onHeartbeatFailed();
                 fireHeartbeatStopped(connection);
             }
+        } else {
+            if (!connection.isHeartBeating()) {
+                logger.warning("Heartbeat is back to healthy for the connection: " + connection);
+                connection.onHeartbeatResumed();
+                fireHeartbeatResumed(connection);
+            }
         }
 
         if (now - connection.lastWriteTimeMillis() > heartbeatInterval) {
@@ -119,12 +125,6 @@ public void onFailure(Throwable t) {
                     }
                 }
             });
-        } else {
-            if (!connection.isHeartBeating()) {
-                logger.warning("Heartbeat is back to healthy for the connection: " + connection);
-                connection.onHeartbeatResumed();
-                fireHeartbeatResumed(connection);
-            }
         }
     }
 
diff --git a/hazelcast-client/src/test/java/com/hazelcast/client/heartbeat/ClientHeartbeatTest.java b/hazelcast-client/src/test/java/com/hazelcast/client/heartbeat/ClientHeartbeatTest.java
index 5b264b6e0c64..9c3a2413a8e2 100644
--- a/hazelcast-client/src/test/java/com/hazelcast/client/heartbeat/ClientHeartbeatTest.java
+++ b/hazelcast-client/src/test/java/com/hazelcast/client/heartbeat/ClientHeartbeatTest.java
@@ -103,7 +103,7 @@ public void heartbeatStopped(Connection connection) {
     }
 
     @Test
-    public void testHeartbeatResumedEvent() throws InterruptedException {
+    public void testHeartbeatResumedEvent() {
         hazelcastFactory.newHazelcastInstance();
         HazelcastInstance client = hazelcastFactory.newHazelcastClient(getClientConfig());
         final HazelcastInstance instance2 = hazelcastFactory.newHazelcastInstance();
@@ -116,22 +116,25 @@ public void testHeartbeatResumedEvent() throws InterruptedException {
         HazelcastClientInstanceImpl clientImpl = getHazelcastClientInstanceImpl(client);
         final ClientConnectionManager connectionManager = clientImpl.getConnectionManager();
         final CountDownLatch countDownLatch = new CountDownLatch(1);
+        final AtomicLong stoppedCount = new AtomicLong();
+        final AtomicLong resumedCount = new AtomicLong();
         connectionManager.addConnectionHeartbeatListener(new ConnectionHeartbeatListener() {
             @Override
             public void heartbeatResumed(Connection connection) {
                 assertEquals(instance2.getCluster().getLocalMember().getAddress(), connection.getEndPoint());
                 countDownLatch.countDown();
+                resumedCount.incrementAndGet();
             }
 
             @Override
             public void heartbeatStopped(Connection connection) {
+                stoppedCount.incrementAndGet();
             }
         });
 
         assertTrueEventually(new AssertTask() {
             @Override
-            public void run()
-                    throws Exception {
+            public void run() {
                 assertNotNull(connectionManager.getActiveConnection(instance2.getCluster().getLocalMember().getAddress()));
             }
         });
@@ -141,6 +144,8 @@ public void run()
         unblockMessagesFromInstance(instance2, client);
 
         assertOpenEventually(countDownLatch);
+        assertEquals(1, resumedCount.get());
+        assertEquals(1, stoppedCount.get());
     }
 
     @Test