Merge pull request #298 from neo4j/1.1-support-member-stop-kill-CC-tests

Support stop/kill of cluster members in CC tests

Author: Zhen Li

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

@@ -62,10 +62,11 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.neo4j.driver.v1.Values.parameters;
 
 public class CausalClusteringIT
 {
-    private static final long DEFAULT_TIMEOUT_MS = 15_000;
+    private static final long DEFAULT_TIMEOUT_MS = 120_000;
 
     @Rule
     public final ClusterRule clusterRule = new ClusterRule();
@@ -293,6 +294,52 @@ public void beginTransactionThrowsForUnreachableBookmark()
         }
     }
 
+    @Test
+    public void shouldHandleGracefulLeaderSwitch() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+        ClusterMember leader = cluster.leader();
+
+        try ( Driver driver = createDriver( leader.getRoutingUri() ) )
+        {
+            Session session1 = driver.session();
+            Transaction tx1 = session1.beginTransaction();
+
+            // gracefully stop current leader to force re-election
+            cluster.stop( leader );
+
+            tx1.run( "CREATE (person:Person {name: {name}, title: {title}})",
+                    parameters( "name", "Webber", "title", "Mr" ) );
+            tx1.success();
+
+            closeAndExpectException( tx1, ClientException.class );
+            closeAndExpectException( session1, ClientException.class );
+
+            String bookmark = inExpirableSession( driver, createSession(), new Function<Session,String>()
+            {
+                @Override
+                public String apply( Session session )
+                {
+                    try ( Transaction tx = session.beginTransaction() )
+                    {
+                        tx.run( "CREATE (person:Person {name: {name}, title: {title}})",
+                                parameters( "name", "Webber", "title", "Mr" ) );
+                        tx.success();
+                    }
+                    return session.lastBookmark();
+                }
+            } );
+
+            try ( Session session2 = driver.session( AccessMode.READ );
+                  Transaction tx2 = session2.beginTransaction( bookmark ) )
+            {
+                Record record = tx2.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                tx2.success();
+                assertEquals( 1, record.get( "count" ).asInt() );
+            }
+        }
+    }
+
     private int executeWriteAndReadThroughBolt( ClusterMember member ) throws TimeoutException, InterruptedException
     {
         try ( Driver driver = createDriver( member.getRoutingUri() ) )
@@ -350,10 +397,12 @@ private <T> T inExpirableSession( Driver driver, Function<Driver,Session> acquir
             {
                 return op.apply( session );
             }
-            catch ( SessionExpiredException e )
+            catch ( SessionExpiredException | ServiceUnavailableException e )
             {
                 // role might have changed; try again;
             }
+
+            Thread.sleep( 500 );
         }
         while ( System.currentTimeMillis() < endTime );
 
@@ -408,4 +457,17 @@ public Void call() throws Exception
         executor.shutdown();
         assertTrue( executor.awaitTermination( 1, TimeUnit.MINUTES ) );
     }
+
+    private static void closeAndExpectException( AutoCloseable closeable, Class<? extends Exception> exceptionClass )
+    {
+        try
+        {
+            closeable.close();
+            fail( "Exception expected" );
+        }
+        catch ( Exception e )
+        {
+            assertThat( e, instanceOf( exceptionClass ) );
+        }
+    }
 }

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

@@ -43,11 +43,13 @@
 public class Cluster
 {
     private static final String ADMIN_USER = "neo4j";
-    private static final int STARTUP_TIMEOUT_SECONDS = 60;
+    private static final int STARTUP_TIMEOUT_SECONDS = 90;
+    private static final int ONLINE_MEMBERS_CHECK_SLEEP_MS = 500;
 
     private final Path path;
     private final String password;
     private final Set<ClusterMember> members;
+    private final Set<ClusterMember> offlineMembers;
 
     public Cluster( Path path, String password )
     {
@@ -59,11 +61,12 @@ private Cluster( Path path, String password, Set<ClusterMember> members )
         this.path = path;
         this.password = password;
         this.members = members;
+        this.offlineMembers = new HashSet<>();
     }
 
     Cluster withMembers( Set<ClusterMember> newMembers ) throws ClusterUnavailableException
     {
-        waitForMembers( newMembers, password );
+        waitForMembersToBeOnline( newMembers, password );
         return new Cluster( path, password, newMembers );
     }
 
@@ -126,6 +129,35 @@ public Set<ClusterMember> readReplicas()
         return membersWithRole( ClusterMemberRole.READ_REPLICA );
     }
 
+    public void start( ClusterMember member )
+    {
+        startNoWait( member );
+        waitForMembersToBeOnline();
+    }
+
+    public void startOfflineMembers()
+    {
+        for ( ClusterMember member : offlineMembers )
+        {
+            startNoWait( member );
+        }
+        waitForMembersToBeOnline();
+    }
+
+    public void stop( ClusterMember member )
+    {
+        removeOfflineMember( member );
+        SharedCluster.stop( member );
+        waitForMembersToBeOnline();
+    }
+
+    public void kill( ClusterMember member )
+    {
+        removeOfflineMember( member );
+        SharedCluster.kill( member );
+        waitForMembersToBeOnline();
+    }
+
     @Override
     public String toString()
     {
@@ -135,6 +167,30 @@ public String toString()
                "}";
     }
 
+    private void addOfflineMember( ClusterMember member )
+    {
+        if ( !offlineMembers.remove( member ) )
+        {
+            throw new IllegalArgumentException( "Cluster member is not offline: " + member );
+        }
+        members.add( member );
+    }
+
+    private void removeOfflineMember( ClusterMember member )
+    {
+        if ( !members.remove( member ) )
+        {
+            throw new IllegalArgumentException( "Unknown cluster member " + member );
+        }
+        offlineMembers.add( member );
+    }
+
+    private void startNoWait( ClusterMember member )
+    {
+        addOfflineMember( member );
+        SharedCluster.start( member );
+    }
+
     private Set<ClusterMember> membersWithRole( ClusterMemberRole role )
     {
         Set<ClusterMember> membersWithRole = new HashSet<>();
@@ -166,41 +222,57 @@ private Set<ClusterMember> membersWithRole( ClusterMemberRole role )
         return membersWithRole;
     }
 
-    private static Set<ClusterMember> waitForMembers( Set<ClusterMember> members, String password )
+    private void waitForMembersToBeOnline()
+    {
+        try
+        {
+            waitForMembersToBeOnline( members, password );
+        }
+        catch ( ClusterUnavailableException e )
+        {
+            throw new RuntimeException( e );
+        }
+    }
+
+    private static void waitForMembersToBeOnline( Set<ClusterMember> members, String password )
             throws ClusterUnavailableException
     {
         if ( members.isEmpty() )
         {
             throw new IllegalArgumentException( "No members to wait for" );
         }
 
-        Set<ClusterMember> offlineMembers = new HashSet<>( members );
+        Set<URI> expectedOnlineUris = extractBoltUris( members );
+        Set<URI> actualOnlineUris = Collections.emptySet();
+
         long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis( STARTUP_TIMEOUT_SECONDS );
+        Throwable error = null;
 
-        try ( Driver driver = createDriver( members, password ) )
+        while ( !expectedOnlineUris.equals( actualOnlineUris ) )
         {
-            while ( !offlineMembers.isEmpty() )
+            sleep( ONLINE_MEMBERS_CHECK_SLEEP_MS );
+            assertDeadlineNotReached( deadline, expectedOnlineUris, actualOnlineUris, error );
+
+            try ( Driver driver = createDriver( members, password );
+                  Session session = driver.session( AccessMode.READ ) )
             {
-                assertDeadlineNotReached( deadline );
+                List<Record> records = findClusterOverview( session );
+                actualOnlineUris = extractBoltUris( records );
+            }
+            catch ( Throwable t )
+            {
+                t.printStackTrace();
 
-                try ( Session session = driver.session( AccessMode.READ ) )
+                if ( error == null )
                 {
-                    List<Record> records = findClusterOverview( session );
-                    for ( Record record : records )
-                    {
-                        URI boltUri = extractBoltUri( record );
-
-                        ClusterMember member = findByBoltUri( boltUri, offlineMembers );
-                        if ( member != null )
-                        {
-                            offlineMembers.remove( member );
-                        }
-                    }
+                    error = t;
+                }
+                else
+                {
+                    error.addSuppressed( t );
                 }
             }
         }
-
-        return members;
     }
 
     private static Driver createDriver( Set<ClusterMember> members, String password )
@@ -243,15 +315,48 @@ private static boolean isCoreMember( Session session )
         return role != ClusterMemberRole.READ_REPLICA;
     }
 
-    private static void assertDeadlineNotReached( long deadline ) throws ClusterUnavailableException
+    private static void assertDeadlineNotReached( long deadline, Set<URI> expectedUris, Set<URI> actualUris,
+            Throwable error ) throws ClusterUnavailableException
     {
         if ( System.currentTimeMillis() > deadline )
         {
-            throw new ClusterUnavailableException(
-                    "Cluster did not become available in " + STARTUP_TIMEOUT_SECONDS + " seconds" );
+            String baseMessage = "Cluster did not become available in " + STARTUP_TIMEOUT_SECONDS + " seconds.\n";
+            String errorMessage = error == null ? "" : "There were errors checking cluster members.\n";
+            String expectedUrisMessage = "Expected online URIs: " + expectedUris + "\n";
+            String actualUrisMessage = "Actual last seen online URIs: " + actualUris + "\n";
+            String message = baseMessage + errorMessage + expectedUrisMessage + actualUrisMessage;
+
+            ClusterUnavailableException clusterUnavailable = new ClusterUnavailableException( message );
+
+            if ( error != null )
+            {
+                clusterUnavailable.addSuppressed( error );
+            }
+
+            throw clusterUnavailable;
         }
     }
 
+    private static Set<URI> extractBoltUris( Set<ClusterMember> members )
+    {
+        Set<URI> uris = new HashSet<>();
+        for ( ClusterMember member : members )
+        {
+            uris.add( member.getBoltUri() );
+        }
+        return uris;
+    }
+
+    private static Set<URI> extractBoltUris( List<Record> records )
+    {
+        Set<URI> uris = new HashSet<>();
+        for ( Record record : records )
+        {
+            uris.add( extractBoltUri( record ) );
+        }
+        return uris;
+    }
+
     private static URI extractBoltUri( Record record )
     {
         List<Object> addresses = record.get( "addresses" ).asList();
@@ -307,4 +412,17 @@ private static ClusterMember randomOf( Set<ClusterMember> members )
         }
         throw new AssertionError();
     }
+
+    private static void sleep( int millis )
+    {
+        try
+        {
+            Thread.sleep( millis );
+        }
+        catch ( InterruptedException e )
+        {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException( e );
+        }
+    }
 }

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

@@ -62,16 +62,31 @@ static String startCluster( Path path )
         return executeCommand( "neoctrl-cluster", "start", path.toString() );
     }
 
+    static String startClusterMember( Path path )
+    {
+        return executeCommand( "neoctrl-start", path.toString() );
+    }
+
     static void stopCluster( Path path )
     {
         executeCommand( "neoctrl-cluster", "stop", path.toString() );
     }
 
+    static void stopClusterMember( Path path )
+    {
+        executeCommand( "neoctrl-stop", path.toString() );
+    }
+
     static void killCluster( Path path )
     {
         executeCommand( "neoctrl-cluster", "stop", "--kill", path.toString() );
     }
 
+    static void killClusterMember( Path path )
+    {
+        executeCommand( "neoctrl-stop", "--kill", path.toString() );
+    }
+
     private static String executeCommand( String... command )
     {
         try

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

@@ -40,7 +40,7 @@ public class ClusterRule extends ExternalResource
     private static final int INITIAL_PORT = 20_000;
 
     // todo: neo4j version should come from environment variables
-    private static final String NEO4J_VERSION = "3.1.0-RC1";
+    private static final String NEO4J_VERSION = "3.1.0";
     // todo: should be possible to configure (dynamically add/remove) cores and read replicas
     private static final int CORE_COUNT = 3;
     private static final int READ_REPLICA_COUNT = 2;
@@ -96,7 +96,9 @@ protected void before() throws Throwable
     @Override
     protected void after()
     {
-        getCluster().deleteData();
+        Cluster cluster = getCluster();
+        cluster.startOfflineMembers();
+        cluster.deleteData();
     }
 
     private static void addShutdownHookToStopCluster()