JAVA-1346: Reset heartbeat only on client reads (not writes).

Alexandre Dutra · Alexandre Dutra · commit 31f9c200979a · 2016-12-21T12:14:18.000+01:00
Contributions by Andrew Tolbert (@tolbertam): Refactor HeartbeatTest to use Scassandra instead of a C* cluster.
diff --git a/changelog/README.md b/changelog/README.md
@@ -4,6 +4,7 @@
 
 - [bug] JAVA-1330: Add un/register for SchemaChangeListener in DelegatingCluster
 - [bug] JAVA-1351: Include Custom Payload in Request.copy.
+- [bug] JAVA-1346: Reset heartbeat only on client reads (not writes).
 
 
 ### 3.0.5
diff --git a/driver-core/src/main/java/com/datastax/driver/core/Connection.java b/driver-core/src/main/java/com/datastax/driver/core/Connection.java
@@ -50,7 +50,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import static com.datastax.driver.core.Message.Response.Type.ERROR;
-import static io.netty.handler.timeout.IdleState.ALL_IDLE;
+import static io.netty.handler.timeout.IdleState.READER_IDLE;
 
 // For LoggingHandler
 //import org.jboss.netty.handler.logging.LoggingHandler;
@@ -1077,7 +1077,7 @@ protected void channelRead0(ChannelHandlerContext ctx, Message.Response response
 
         @Override
         public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-            if (!isClosed() && evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state() == ALL_IDLE) {
+            if (!isClosed() && evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state() == READER_IDLE) {
                 logger.debug("{} was inactive for {} seconds, sending heartbeat", Connection.this, factory.configuration.getPoolingOptions().getHeartbeatIntervalSeconds());
                 write(HEARTBEAT_CALLBACK);
             }
@@ -1406,7 +1406,7 @@ private static class Initializer extends ChannelInitializer<SocketChannel> {
             this.sslOptions = sslOptions;
             this.nettyOptions = nettyOptions;
             this.codecRegistry = codecRegistry;
-            this.idleStateHandler = new IdleStateHandler(0, 0, heartBeatIntervalSeconds);
+            this.idleStateHandler = new IdleStateHandler(heartBeatIntervalSeconds, 0, 0);
         }
 
         @Override
diff --git a/driver-core/src/test/java/com/datastax/driver/core/HeartbeatTest.java b/driver-core/src/test/java/com/datastax/driver/core/HeartbeatTest.java
@@ -15,18 +15,26 @@
  */
 package com.datastax.driver.core;
 
+import com.google.common.util.concurrent.Uninterruptibles;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
 import static com.datastax.driver.core.Assertions.assertThat;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.fail;
+import static org.scassandra.http.client.PrimingRequest.queryBuilder;
+import static org.scassandra.http.client.PrimingRequest.then;
 
-@CCMConfig(createCluster = false)
-public class HeartbeatTest extends CCMTestsSupport {
+public class HeartbeatTest extends ScassandraTestBase {
 
+    static org.slf4j.Logger logger = LoggerFactory.getLogger(HeartbeatTest.class);
     Logger connectionLogger = Logger.getLogger(Connection.class);
     MemoryAppender logs;
     Level originalLevel;
@@ -56,12 +64,12 @@ public void stopCapturingLogs() {
      */
     @Test(groups = "long")
     public void should_send_heartbeat_when_connection_is_inactive() throws InterruptedException {
-        Cluster cluster = register(Cluster.builder()
-                .addContactPoints(getContactPoints().get(0))
-                .withPort(ccm().getBinaryPort())
-                .withPoolingOptions(new PoolingOptions()
-                        .setHeartbeatIntervalSeconds(3))
-                .build());
+        Cluster cluster = Cluster.builder()
+                .addContactPoints(hostAddress.getAddress())
+                .withPort(scassandra.getBinaryPort())
+                .withProtocolVersion(ProtocolVersion.V2)
+                .withPoolingOptions(new PoolingOptions().setHeartbeatIntervalSeconds(3))
+                .build();
 
         try {
             // Don't create any session, only the control connection will be established
@@ -103,6 +111,134 @@ public void should_send_heartbeat_when_connection_is_inactive() throws Interrupt
         }
     }
 
+    /**
+     * Verifies that there exists a line in logs that matches pattern.
+     *
+     * @param logs    Captured log entries.
+     * @param pattern Pattern to match on individual lines.
+     * @return
+     */
+    private void assertLineMatches(String logs, Pattern pattern) {
+        String lines[] = logs.split("\\r?\\n");
+        for (String line : lines) {
+            if (pattern.matcher(line).matches()) {
+                return;
+            }
+        }
+        fail("Expecting: [" + logs + "] to contain " + pattern);
+    }
+
+    /**
+     * Verifies that no line in logs matches pattern.
+     *
+     * @param logs    Captured log entries.
+     * @param pattern Pattern to match on individual lines.
+     */
+    private void assertNoLineMatches(String logs, Pattern pattern) {
+        String lines[] = logs.split("\\r?\\n");
+        for (String line : lines) {
+            if (pattern.matcher(line).matches()) {
+                fail("Expecting: [" + logs + "] not to contain " + pattern);
+            }
+        }
+    }
+
+    /**
+     * Ensures that a heartbeat message is sent after the configured heartbeat interval of idle time when no data is
+     * received on a connection even though are successful writes on the socket.
+     *
+     * @test_category connection:heartbeat
+     * @expected_result heartbeat is sent after heartbeat interval (3) seconds of idle time.
+     * @jira_ticket JAVA-1346
+     * @since 3.0.6, 3.1.3
+     */
+    @Test(groups = "long")
+    public void should_send_heartbeat_when_requests_being_written_but_nothing_received() throws Exception {
+        Cluster cluster = Cluster.builder()
+                .addContactPoints(hostAddress.getAddress())
+                .withPort(scassandra.getBinaryPort())
+                .withProtocolVersion(ProtocolVersion.V2)
+                .withPoolingOptions(new PoolingOptions().setHeartbeatIntervalSeconds(3).setConnectionsPerHost(HostDistance.LOCAL, 1, 1))
+                .build();
+
+        // Prime 'ping' to never return a response this is a way to create outgoing traffic
+        // without receiving anything inbound.
+        scassandra.primingClient()
+                .prime(queryBuilder().withQuery("ping").withThen(then().withFixedDelay(8675309999L)));
+
+        // Thread that will submit queries that get no response repeatedly.
+        Thread submitter = null;
+        try {
+            // Don't create any session, only the control connection will be established
+            cluster.init();
+
+            // Find the connection in the connection pool.
+            SessionManager session = (SessionManager) cluster.connect();
+            Host host = TestUtils.findHost(cluster, 1);
+            Connection connection = session.pools.get(host).connections.get(0);
+
+            // Extract connection name from toString implementation.
+            String connectionName = connection.toString()
+                    .replaceAll("\\-", "\\\\-") // Replace - with \- so its properly escaped as a regex.
+                    .replaceAll("Connection\\[\\/", "") // Replace first part of toString (Connection[
+                    .replaceAll("\\, inFlight.*", ""); // Replace everything after ',inFlight'
+
+            // Define patterns that check for whether or not heartbeats are sent / received on a given connection.
+            Pattern heartbeatSentPattern = Pattern.compile(".*" + connectionName + ".*sending heartbeat");
+            Pattern heartbeatReceivedPattern = Pattern.compile(".*" + connectionName + ".*heartbeat query succeeded");
+            logger.debug("Heartbeat pattern is {}", heartbeatSentPattern);
+
+            // Start query submission thread.
+            submitter = new Thread(new QuerySubmitter(session));
+            submitter.start();
+
+            for (int i = 0; i < 5; i++) {
+                session.execute("bar");
+                SECONDS.sleep(1);
+            }
+
+            // Should be no heartbeats sent on pooled connection since we had successful requests.
+            String log = logs.getNext();
+            assertNoLineMatches(log, heartbeatSentPattern);
+
+            int inFlight = connection.inFlight.get();
+            assertThat(inFlight).isGreaterThan(0);
+
+            // Ensure heartbeat is sent after no received data, even though we have inflight requests (JAVA-1346).
+            SECONDS.sleep(4);
+            // Verify more requests were sent over this time period.
+            assertThat(connection.inFlight.get()).isGreaterThan(inFlight);
+            log = logs.getNext();
+            // Heartbeat should have been sent and received.
+            assertLineMatches(log, heartbeatSentPattern);
+            assertLineMatches(log, heartbeatReceivedPattern);
+        } finally {
+            // interrupt thread so it stops submitting queries.
+            if (submitter != null) {
+                submitter.interrupt();
+            }
+            cluster.close();
+        }
+    }
+
+    private static class QuerySubmitter implements Runnable {
+
+        private final Session session;
+
+        QuerySubmitter(Session session) {
+            this.session = session;
+        }
+
+        @Override
+        public void run() {
+            while (!Thread.currentThread().isInterrupted()) {
+                logger.debug("Sending ping, for which we expect no response");
+                session.executeAsync("ping");
+                Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+            }
+        }
+    }
+
     /**
      * Ensures that a heartbeat message is not sent if the configured heartbeat interval is 0.
      * <p/>
@@ -116,12 +252,13 @@ public void should_send_heartbeat_when_connection_is_inactive() throws Interrupt
      */
     @Test(groups = "long")
     public void should_not_send_heartbeat_when_disabled() throws InterruptedException {
-        Cluster cluster = register(Cluster.builder()
-                .addContactPoints(getContactPoints().get(0))
-                .withPort(ccm().getBinaryPort())
+        Cluster cluster = Cluster.builder()
+                .addContactPoints(hostAddress.getAddress())
+                .withPort(scassandra.getBinaryPort())
                 .withPoolingOptions(new PoolingOptions()
                         .setHeartbeatIntervalSeconds(0))
-                .build());
+                .withProtocolVersion(ProtocolVersion.V2)
+                .build();
 
         try {
             // Don't create any session, only the control connection will be established
