Closing all connections when heartbeat timeout

This was actually addressed in following pr.
Members and clients disconnect the connection as soon as
heartbeat timeout is detected.

hazelcast#12765

This pr fixes member side heartbeat monitor. It was not closing
the connections if the connection is not an owner connection.

fixes hazelcast#14094

hazelcast-client/src/test/java/com/hazelcast/client/heartbeat/ClientHeartbeatTest.java

@@ -79,7 +79,7 @@ public void cleanup() {
     }
 
     @Test
-    public void testConnectionClosed_whenHeartbeatStopped() throws InterruptedException {
+    public void testOwnerConnectionClosed_whenHeartbeatStopped() throws InterruptedException {
         HazelcastInstance instance = hazelcastFactory.newHazelcastInstance();
         HazelcastInstance client = hazelcastFactory.newHazelcastClient(getClientConfig());
 
@@ -102,6 +102,39 @@ public void connectionRemoved(Connection connection) {
         assertOpenEventually(countDownLatch);
     }
 
+    @Test
+    public void testNonOwnerConnectionClosed_whenHeartbeatStopped() throws InterruptedException {
+        hazelcastFactory.newHazelcastInstance();
+        HazelcastInstance client = hazelcastFactory.newHazelcastClient(getClientConfig());
+        HazelcastInstance instance = hazelcastFactory.newHazelcastInstance();
+
+        HazelcastClientInstanceImpl clientImpl = getHazelcastClientInstanceImpl(client);
+        final ClientConnectionManager connectionManager = clientImpl.getConnectionManager();
+
+        assertTrueEventually(new AssertTask() {
+            @Override
+            public void run() throws Exception {
+                assertEquals(2, connectionManager.getActiveConnections().size());
+            }
+        });
+
+        final CountDownLatch countDownLatch = new CountDownLatch(1);
+        connectionManager.addConnectionListener(new ConnectionListener() {
+            @Override
+            public void connectionAdded(Connection connection) {
+
+            }
+
+            @Override
+            public void connectionRemoved(Connection connection) {
+                countDownLatch.countDown();
+            }
+        });
+
+        blockMessagesFromInstance(instance, client);
+        assertOpenEventually(countDownLatch);
+    }
+
     @Test
     public void testInvocation_whenHeartbeatStopped() throws InterruptedException {
         hazelcastFactory.newHazelcastInstance();

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

@@ -92,9 +92,9 @@ private void cleanupEndpointsWithDeadConnections() {
     }
 
     private void monitor(ClientEndpoint clientEndpoint) {
-        // C++ client sends heartbeat over its non-owner connections
-        // For other client types, disregard non-owner connections for heartbeat monitoring purposes
-        if (clientEndpoint.isOwnerConnection() == ClientType.CPP.equals(clientEndpoint.getClientType())) {
+        // C++ client does not send heartbeat over its owner connection
+        // We are skipping checking heartbeat for cpp owner connection
+        if (clientEndpoint.isOwnerConnection() && ClientType.CPP.equals(clientEndpoint.getClientType())) {
             return;
         }