Unflake couple CC integration tests

New test 'shouldKeepOperatingWhenConnectionsBreak' was flaky because
it's reads sometimes failed with `EntityNotFoundException`. Test itself
does not remove any nodes. It stepped on transactions created by
previous data cleanup that is executed between tests. That cleanup
did not actually wait for all cluster members to observe removals.
This commit makes cleanup procedure use bookmark and count nodes on
all cluster members.

Author: lutovich

driver/src/test/java/org/neo4j/driver/v1/integration/CausalClusteringIT.java

@@ -632,6 +632,7 @@ public void shouldRediscoverWhenConnectionsToAllCoresBreak()
     @Test
     public void shouldKeepOperatingWhenConnectionsBreak() throws Exception
     {
+        long testRunTimeMs = MINUTES.toMillis( 1 );
         String label = "Person";
         String property = "name";
         String value = "Tony Stark";
@@ -641,8 +642,13 @@ public void shouldKeepOperatingWhenConnectionsBreak() throws Exception
         AtomicBoolean stop = new AtomicBoolean();
         executor = newExecutor();
 
+        Config config = Config.build()
+                .withLogging( DEV_NULL_LOGGING )
+                .withMaxTransactionRetryTime( testRunTimeMs, MILLISECONDS )
+                .toConfig();
+
         try ( Driver driver = driverFactory.newInstance( cluster.leader().getRoutingUri(), clusterRule.getDefaultAuthToken(),
-                defaultRoutingSettings(), RetrySettings.DEFAULT, configWithoutLogging() ) )
+                defaultRoutingSettings(), RetrySettings.DEFAULT, config ) )
         {
             List<Future<?>> results = new ArrayList<>();
 
@@ -665,7 +671,7 @@ public void shouldKeepOperatingWhenConnectionsBreak() throws Exception
                 {
                     connection.setNextRunError( new ServiceUnavailableException( "Unable to execute query" ) );
                 }
-                SECONDS.sleep( 5 ); // sleep a bit to allow readers and writers to progress
+                SECONDS.sleep( 10 ); // sleep a bit to allow readers and writers to progress
             }
             stop.set( true );
 

driver/src/test/java/org/neo4j/driver/v1/util/TestUtil.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2002-2018 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.v1.util;
+
+import org.neo4j.driver.v1.Driver;
+import org.neo4j.driver.v1.Session;
+import org.neo4j.driver.v1.StatementResult;
+import org.neo4j.driver.v1.Transaction;
+import org.neo4j.driver.v1.TransactionWork;
+
+public final class TestUtil
+{
+    private TestUtil()
+    {
+    }
+
+    public static long countNodes( Driver driver, String bookmark )
+    {
+        try ( Session session = driver.session( bookmark ) )
+        {
+            return session.readTransaction( new TransactionWork<Long>()
+            {
+                @Override
+                public Long execute( Transaction tx )
+                {
+                    return tx.run( "MATCH (n) RETURN count(n)" ).single().get( 0 ).asLong();
+                }
+            } );
+        }
+    }
+
+    public static String cleanDb( Driver driver )
+    {
+        try ( Session session = driver.session() )
+        {
+            cleanDb( session );
+            return session.lastBookmark();
+        }
+    }
+
+    private static void cleanDb( Session session )
+    {
+        int nodesDeleted;
+        do
+        {
+            nodesDeleted = deleteBatchOfNodes( session );
+        }
+        while ( nodesDeleted > 0 );
+    }
+
+    private static int deleteBatchOfNodes( Session session )
+    {
+        return session.writeTransaction( new TransactionWork<Integer>()
+        {
+            @Override
+            public Integer execute( Transaction tx )
+            {
+                StatementResult result = tx.run( "MATCH (n) WITH n LIMIT 10000 DETACH DELETE n RETURN count(n)" );
+                return result.single().get( 0 ).asInt();
+            }
+        } );
+    }
+}

driver/src/test/java/org/neo4j/driver/v1/util/cc/Cluster.java

@@ -29,7 +29,6 @@
 import java.util.concurrent.TimeUnit;
 
 import org.neo4j.driver.internal.net.BoltServerAddress;
-import org.neo4j.driver.internal.util.Consumer;
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.AuthTokens;
 import org.neo4j.driver.v1.Config;
@@ -38,8 +37,11 @@
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.Session;
 import org.neo4j.driver.v1.StatementResult;
+import org.neo4j.driver.v1.util.TestUtil;
 
 import static java.util.Collections.unmodifiableSet;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.neo4j.driver.internal.util.Iterables.single;
 import static org.neo4j.driver.v1.Config.TrustStrategy.trustAllCertificates;
 
@@ -80,26 +82,23 @@ public Path getPath()
 
     public void deleteData()
     {
-        leaderTx( new Consumer<Session>()
+        // execute write query to remove all nodes and retreive bookmark
+        String bookmark;
+        try ( Driver driver = createDriver( leader().getBoltUri(), password ) )
         {
-            @Override
-            public void accept( Session session )
-            {
-                session.run( "MATCH (n) DETACH DELETE n" ).consume();
-            }
-        } );
-    }
+            bookmark = TestUtil.cleanDb( driver );
+            assertNotNull( "Cleanup of the database did not produce a bookmark", bookmark );
+        }
 
-    public ClusterMember leaderTx( Consumer<Session> tx )
-    {
-        ClusterMember leader = leader();
-        try ( Driver driver = createDriver( leader.getBoltUri(), password );
-              Session session = driver.session() )
+        // ensure that every cluster member is up-to-date and contains no nodes
+        for ( ClusterMember member : members )
         {
-            tx.accept( session );
+            try ( Driver driver = createDriver( member.getBoltUri(), password ) )
+            {
+                long nodeCount = TestUtil.countNodes( driver, bookmark );
+                assertEquals( "Not all nodes have been deleted. " + nodeCount + " still there somehow ", 0L, nodeCount );
+            }
         }
-
-        return leader;
     }
 
     public Set<ClusterMember> members()
@@ -314,13 +313,13 @@ private static Driver createDriver( Set<ClusterMember> members, String password
 
     private static List<Record> findClusterOverview( Session session )
     {
-        StatementResult result = session.run( "call dbms.cluster.overview" );
+        StatementResult result = session.run( "CALL dbms.cluster.overview()" );
         return result.list();
     }
 
     private static boolean isCoreMember( Session session )
     {
-        Record record = single( session.run( "call dbms.cluster.role" ).list() );
+        Record record = single( session.run( "CALL dbms.cluster.role()" ).list() );
         ClusterMemberRole role = extractRole( record );
         return role != ClusterMemberRole.READ_REPLICA;
     }

driver/src/test/java/org/neo4j/driver/v1/util/cc/ClusterRule.java

@@ -133,7 +133,10 @@ public void run()
             {
                 try
                 {
-                    SharedCluster.kill();
+                    if ( SharedCluster.exists() )
+                    {
+                        SharedCluster.kill();
+                    }
                 }
                 catch ( Throwable t )
                 {