Merge pull request #289 from lutovich/1.1-causal-cluster-it

Basic infrastructure for causal cluster integration tests

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/util/Iterables.java

@@ -52,6 +52,21 @@ public static <T> List<T> asList( Iterable<T> it )
         return list;
     }
 
+    public static <T> T single( Iterable<T> it )
+    {
+        Iterator<T> iterator = it.iterator();
+        if ( !iterator.hasNext() )
+        {
+            throw new IllegalArgumentException( "Given iterable is empty" );
+        }
+        T result = iterator.next();
+        if ( iterator.hasNext() )
+        {
+            throw new IllegalArgumentException( "Given iterable contains more than one element: " + it );
+        }
+        return result;
+    }
+
     public static Map<String, String> map( String ... alternatingKeyValue )
     {
         Map<String, String> out = new HashMap<>();

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

@@ -0,0 +1,268 @@
+/*
+ * Copyright (c) 2002-2016 "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.integration;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.net.URI;
+import java.util.concurrent.TimeoutException;
+
+import org.neo4j.driver.internal.logging.DevNullLogger;
+import org.neo4j.driver.v1.AccessMode;
+import org.neo4j.driver.v1.Config;
+import org.neo4j.driver.v1.Driver;
+import org.neo4j.driver.v1.GraphDatabase;
+import org.neo4j.driver.v1.Logger;
+import org.neo4j.driver.v1.Logging;
+import org.neo4j.driver.v1.Record;
+import org.neo4j.driver.v1.Session;
+import org.neo4j.driver.v1.Transaction;
+import org.neo4j.driver.v1.Values;
+import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
+import org.neo4j.driver.v1.exceptions.SessionExpiredException;
+import org.neo4j.driver.v1.util.Function;
+import org.neo4j.driver.v1.util.cc.Cluster;
+import org.neo4j.driver.v1.util.cc.ClusterMember;
+import org.neo4j.driver.v1.util.cc.ClusterRule;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+public class CausalClusteringIT
+{
+    private static final long DEFAULT_TIMEOUT_MS = 15_000;
+
+    @Rule
+    public final ClusterRule clusterRule = new ClusterRule();
+
+    @Test
+    public void shouldExecuteReadAndWritesWhenDriverSuppliedWithAddressOfLeader() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+
+        int count = executeWriteAndReadThroughBolt( cluster.leader() );
+
+        assertEquals( 1, count );
+    }
+
+    @Test
+    public void shouldExecuteReadAndWritesWhenDriverSuppliedWithAddressOfFollower() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+
+        int count = executeWriteAndReadThroughBolt( cluster.anyFollower() );
+
+        assertEquals( 1, count );
+    }
+
+    @Test
+    public void sessionCreationShouldFailIfCallingDiscoveryProcedureOnEdgeServer() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+
+        ClusterMember readReplica = cluster.anyReadReplica();
+        try
+        {
+            createDriver( readReplica.getRoutingUri() );
+            fail( "Should have thrown an exception using a read replica address for routing" );
+        }
+        catch ( ServiceUnavailableException ex )
+        {
+            assertEquals( "Could not perform discovery. No routing servers available.", ex.getMessage() );
+        }
+    }
+
+    // Ensure that Bookmarks work with single instances using a driver created using a bolt[not+routing] URI.
+    @Test
+    public void bookmarksShouldWorkWithDriverPinnedToSingleServer() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+        ClusterMember leader = cluster.leader();
+
+        try ( Driver driver = createDriver( leader.getBoltUri() ) )
+        {
+            String bookmark = inExpirableSession( driver, createSession(), new Function<Session,String>()
+            {
+                @Override
+                public String apply( Session session )
+                {
+                    try ( Transaction tx = session.beginTransaction() )
+                    {
+                        tx.run( "CREATE (p:Person {name: {name} })", Values.parameters( "name", "Alistair" ) );
+                        tx.success();
+                    }
+
+                    return session.lastBookmark();
+                }
+            } );
+
+            assertNotNull( bookmark );
+
+            try ( Session session = driver.session();
+                  Transaction tx = session.beginTransaction( bookmark ) )
+            {
+                Record record = tx.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                assertEquals( 1, record.get( "count" ).asInt() );
+                tx.success();
+            }
+        }
+    }
+
+    @Test
+    public void shouldUseBookmarkFromAReadSessionInAWriteSession() throws Exception
+    {
+        Cluster cluster = clusterRule.getCluster();
+        ClusterMember leader = cluster.leader();
+
+        try ( Driver driver = createDriver( leader.getBoltUri() ) )
+        {
+            inExpirableSession( driver, createWritableSession(), new Function<Session,Void>()
+            {
+                @Override
+                public Void apply( Session session )
+                {
+                    session.run( "CREATE (p:Person {name: {name} })", Values.parameters( "name", "Jim" ) );
+                    return null;
+                }
+            } );
+
+            final String bookmark;
+            try ( Session session = driver.session( AccessMode.READ ) )
+            {
+                try ( Transaction tx = session.beginTransaction() )
+                {
+                    tx.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                    tx.success();
+                }
+
+                bookmark = session.lastBookmark();
+            }
+
+            assertNotNull( bookmark );
+
+            inExpirableSession( driver, createWritableSession(), new Function<Session,Void>()
+            {
+                @Override
+                public Void apply( Session session )
+                {
+                    try ( Transaction tx = session.beginTransaction( bookmark ) )
+                    {
+                        tx.run( "CREATE (p:Person {name: {name} })", Values.parameters( "name", "Alistair" ) );
+                        tx.success();
+                    }
+
+                    return null;
+                }
+            } );
+
+            try ( Session session = driver.session() )
+            {
+                Record record = session.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                assertEquals( 2, record.get( "count" ).asInt() );
+            }
+        }
+    }
+
+    private int executeWriteAndReadThroughBolt( ClusterMember member ) throws TimeoutException, InterruptedException
+    {
+        try ( Driver driver = createDriver( member.getRoutingUri() ) )
+        {
+            return inExpirableSession( driver, createWritableSession(), executeWriteAndRead() );
+        }
+    }
+
+    private Function<Driver,Session> createSession()
+    {
+        return new Function<Driver,Session>()
+        {
+            @Override
+            public Session apply( Driver driver )
+            {
+                return driver.session();
+            }
+        };
+    }
+
+    private Function<Driver,Session> createWritableSession()
+    {
+        return new Function<Driver,Session>()
+        {
+            @Override
+            public Session apply( Driver driver )
+            {
+                return driver.session( AccessMode.WRITE );
+            }
+        };
+    }
+
+    private Function<Session,Integer> executeWriteAndRead()
+    {
+        return new Function<Session,Integer>()
+        {
+            @Override
+            public Integer apply( Session session )
+            {
+                session.run( "MERGE (n:Person {name: 'Jim'})" ).consume();
+                Record record = session.run( "MATCH (n:Person) RETURN COUNT(*) AS count" ).next();
+                return record.get( "count" ).asInt();
+            }
+        };
+    }
+
+    private <T> T inExpirableSession( Driver driver, Function<Driver,Session> acquirer, Function<Session,T> op )
+            throws TimeoutException, InterruptedException
+    {
+        long endTime = System.currentTimeMillis() + DEFAULT_TIMEOUT_MS;
+
+        do
+        {
+            try ( Session session = acquirer.apply( driver ) )
+            {
+                return op.apply( session );
+            }
+            catch ( SessionExpiredException e )
+            {
+                // role might have changed; try again;
+            }
+        }
+        while ( System.currentTimeMillis() < endTime );
+
+        throw new TimeoutException( "Transaction did not succeed in time" );
+    }
+
+    private Driver createDriver( URI boltUri )
+    {
+        Logging devNullLogging = new Logging()
+        {
+            @Override
+            public Logger getLog( String name )
+            {
+                return DevNullLogger.DEV_NULL_LOGGER;
+            }
+        };
+
+        Config config = Config.build()
+                .withLogging( devNullLogging )
+                .toConfig();
+
+        return GraphDatabase.driver( boltUri, clusterRule.getDefaultAuthToken(), config );
+    }
+}

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

@@ -159,14 +159,7 @@ public void restartNeo4j(Neo4jSettings neo4jSettings) throws IOException
     private int runCommand( String... cmd ) throws IOException
     {
         ProcessBuilder pb = new ProcessBuilder().inheritIO();
-        Map<String,String> env = System.getenv();
-        pb.environment().put( "JAVA_HOME",
-                // This driver is built to work with multiple java versions.
-                // Neo4j, however, works with a specific version of Java. This allows
-                // specifying which Java version to use for Neo4j separately from which
-                // version to use for the driver tests.
-                env.containsKey( "NEO4J_JAVA" ) ? env.get( "NEO4J_JAVA" ) :
-                System.getProperties().getProperty( "java.home" ) );
+        ProcessEnvConfigurator.configure( pb );
         if( NEORUN_START_ARGS != null )
         {
             // overwrite the env var in the sub process if the system property is specified

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2002-2016 "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 java.util.Map;
+
+public final class ProcessEnvConfigurator
+{
+    /**
+     * Name of environment variable used by the Neo4j database.
+     */
+    private static final String JAVA_HOME = "JAVA_HOME";
+    /**
+     * Name of environment variable to be used for the Neo4j database, defined by the build system.
+     */
+    private static final String NEO4J_JAVA = "NEO4J_JAVA";
+
+    private ProcessEnvConfigurator()
+    {
+    }
+
+    public static void configure( ProcessBuilder processBuilder )
+    {
+        processBuilder.environment().put( JAVA_HOME, determineJavaHome() );
+    }
+
+    /**
+     * This driver is built to work with multiple java versions. Neo4j, however, works with a specific version of
+     * Java. This allows specifying which Java version to use for Neo4j separately from which version to use for
+     * the driver tests.
+     * <p>
+     * This method determines which java home to use based on present environment variables.
+     *
+     * @return path to the java home.
+     */
+    private static String determineJavaHome()
+    {
+        Map<String,String> environment = System.getenv();
+
+        String definedJava = environment.get( NEO4J_JAVA );
+        if ( definedJava != null )
+        {
+            return definedJava;
+        }
+
+        return System.getProperties().getProperty( "java.home" );
+    }
+}