For routing driver, by default, get the routing table for default-database.

As this makes the routing table map easier for server version lower than 4.0.
Otherwise, if we default to get routing table for system_db, we need to detect if it is server version lower than 4.0. If so, we can only get routing table for default_db, and then we need to reset the key in routing table map to be default_db.

TODO: Removing stale routing table from routing table map.
Also missing any integration tests for 4.0 routing driver.

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingTableFactory.java renamed to driver/src/main/java/org/neo4j/driver/exceptions/RoutingException.java

@@ -16,9 +16,22 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.neo4j.driver.internal.cluster;
+package org.neo4j.driver.exceptions;
 
-public interface RoutingTableFactory
+/**
+ * A routing error indicate a fatal problem to obtain routing tables such as the routing table for a specified database does not exist.
+ * This exception should not be retried.
+ * @since 2.0
+ */
+public class RoutingException extends Neo4jException
 {
-    RoutingTable newInstance( String databaseName );
+    public RoutingException( String message )
+    {
+        super( message );
+    }
+
+    public RoutingException( String message, Throwable throwable )
+    {
+        super( message, throwable);
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/DirectConnectionProvider.java

@@ -27,7 +27,7 @@
 import org.neo4j.driver.internal.spi.ConnectionProvider;
 
 import static org.neo4j.driver.AccessMode.READ;
-import static org.neo4j.driver.internal.messaging.request.MultiDatabaseUtil.SYSTEM_DB_NAME;
+import static org.neo4j.driver.internal.messaging.request.MultiDatabaseUtil.ABSENT_DB_NAME;
 
 /**
  * Simple {@link ConnectionProvider connection provider} that obtains connections form the given pool only for
@@ -53,8 +53,9 @@ public CompletionStage<Connection> acquireConnection( String databaseName, Acces
     @Override
     public CompletionStage<Void> verifyConnectivity()
     {
-        // we verify the connection by establishing the connection to the default database
-        return acquireConnection( SYSTEM_DB_NAME, READ ).thenCompose( Connection::release );
+        // We verify the connection by establishing a connection with the remote server specified by the address.
+        // Database name will be ignored as no query is run in this connection and the connection is released immediately.
+        return acquireConnection( ABSENT_DB_NAME, READ ).thenCompose( Connection::release );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -127,7 +127,6 @@ public void retainAll( Set<BoltServerAddress> addressesToRetain )
                 {
                     // address is not present in updated routing table and has no active connections
                     // it's now safe to terminate corresponding connection pool and forget about it
-
                     ChannelPool pool = pools.remove( address );
                     if ( pool != null )
                     {

driver/src/main/java/org/neo4j/driver/internal/cluster/Rediscovery.java

@@ -30,19 +30,20 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
 
+import org.neo4j.driver.Logger;
+import org.neo4j.driver.exceptions.SecurityException;
+import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.ConnectionPool;
 import org.neo4j.driver.internal.util.Futures;
-import org.neo4j.driver.Logger;
-import org.neo4j.driver.exceptions.SecurityException;
-import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.net.ServerAddressResolver;
 
 import static java.lang.String.format;
 import static java.util.Collections.emptySet;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.stream.Collectors.toList;
+import static org.neo4j.driver.internal.util.ErrorUtil.isRoutingError;
 import static org.neo4j.driver.internal.util.Futures.completedWithNull;
 import static org.neo4j.driver.internal.util.Futures.failedFuture;
 
@@ -253,12 +254,11 @@ private CompletionStage<ClusterComposition> lookupOnRouter( BoltServerAddress ro
     private ClusterComposition handleRoutingProcedureError( Throwable error, RoutingTable routingTable,
             BoltServerAddress routerAddress )
     {
-        if ( error instanceof SecurityException )
+        if ( error instanceof SecurityException || isRoutingError( error ) )
         {
-            // auth error happened, terminate the discovery procedure immediately
+            // auth error or routing error happened, terminate the discovery procedure immediately
             throw new CompletionException( error );
         }
-        else
         {
             // connection turned out to be broken
             logger.info( format( "Failed to connect to routing server '%s'.", routerAddress ), error );

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingProcedureRunner.java

@@ -19,6 +19,7 @@
 package org.neo4j.driver.internal.cluster;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 
@@ -28,6 +29,7 @@
 import org.neo4j.driver.TransactionConfig;
 import org.neo4j.driver.async.StatementResultCursor;
 import org.neo4j.driver.exceptions.ClientException;
+import org.neo4j.driver.exceptions.RoutingException;
 import org.neo4j.driver.internal.BookmarksHolder;
 import org.neo4j.driver.internal.async.connection.DirectConnection;
 import org.neo4j.driver.internal.spi.Connection;
@@ -37,12 +39,10 @@
 import static org.neo4j.driver.Values.parameters;
 import static org.neo4j.driver.internal.messaging.request.MultiDatabaseUtil.ABSENT_DB_NAME;
 import static org.neo4j.driver.internal.messaging.request.MultiDatabaseUtil.SYSTEM_DB_NAME;
-import static org.neo4j.driver.internal.util.ServerVersion.v3_2_0;
 import static org.neo4j.driver.internal.util.ServerVersion.v4_0_0;
 
 public class RoutingProcedureRunner
 {
-    static final String GET_SERVERS = "dbms.cluster.routing.getServers";
     static final String ROUTING_CONTEXT = "context";
     static final String GET_ROUTING_TABLE = "dbms.cluster.routing.getRoutingTable({" + ROUTING_CONTEXT + "})";
     static final String DATABASE_NAME = "database";
@@ -60,7 +60,8 @@ public CompletionStage<RoutingProcedureResponse> run( CompletionStage<Connection
         return connectionStage.thenCompose( connection ->
         {
             ServerVersion serverVersion = connection.serverVersion();
-            DirectConnection delegate = new DirectConnection( connection, selectDatabase( serverVersion ), AccessMode.WRITE );
+            // As the connection can connect to any router (a.k.a. any core members), this connection strictly speaking is a read connection.
+            DirectConnection delegate = new DirectConnection( connection, SYSTEM_DB_NAME, AccessMode.READ );
             Statement procedure = procedureStatement( serverVersion, databaseName );
             return runProcedure( delegate, procedure )
                     .thenCompose( records -> releaseConnection( delegate, records ) )
@@ -70,7 +71,7 @@ public CompletionStage<RoutingProcedureResponse> run( CompletionStage<Connection
 
     CompletionStage<List<Record>> runProcedure( Connection connection, Statement procedure )
     {
-        return connection.protocol() // this line fails if database name is provided in BOLT versions that do not support database name.
+        return connection.protocol()
                 .runInAutoCommitTransaction( connection, procedure, BookmarksHolder.NO_OP, TransactionConfig.empty(), true )
                 .asyncResult().thenCompose( StatementResultCursor::listAsync );
     }
@@ -86,29 +87,16 @@ private Statement procedureStatement( ServerVersion serverVersion, String databa
             return new Statement( "CALL " + MULTI_DB_GET_ROUTING_TABLE,
                     parameters( ROUTING_CONTEXT, context.asMap(), DATABASE_NAME, databaseName ) );
         }
-        else if ( serverVersion.greaterThanOrEqual( v3_2_0 ) )
+        else
         {
+            if ( !Objects.equals( ABSENT_DB_NAME, databaseName ) )
+            {
+                throw new RoutingException( String.format( "Refreshing routing table for multi-databases is not supported in server version lower than 4.0. " +
+                        "Current server version: %s. Database name: `%s`", serverVersion, databaseName ) );
+            }
             return new Statement( "CALL " + GET_ROUTING_TABLE,
                     parameters( ROUTING_CONTEXT, context.asMap() ) );
         }
-        else
-        {
-            return new Statement( "CALL " + GET_SERVERS );
-        }
-    }
-
-    private String selectDatabase( ServerVersion serverVersion )
-    {
-        if ( serverVersion.greaterThanOrEqual( v4_0_0 ) )
-        {
-            // Routing procedure will be called on the system database
-            return SYSTEM_DB_NAME;
-        }
-        else
-        {
-            // For lower bolt versions, there is no system database, so we should just run on the "default" database
-            return ABSENT_DB_NAME;
-        }
     }
 
     private CompletionStage<List<Record>> releaseConnection( Connection connection, List<Record> records )

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingTableFactoryImpl.java

@@ -34,6 +34,7 @@
 public class RoutingTableHandler implements RoutingErrorHandler
 {
     private final RoutingTable routingTable;
+    private final String databaseName;
     private final RoutingTables routingTables;
     private CompletableFuture<RoutingTable> refreshRoutingTableFuture;
     private final ConnectionPool connectionPool;
@@ -43,6 +44,7 @@ public class RoutingTableHandler implements RoutingErrorHandler
     public RoutingTableHandler( RoutingTable routingTable, Rediscovery rediscovery, ConnectionPool connectionPool, RoutingTables routingTables, Logger log )
     {
         this.routingTable = routingTable;
+        this.databaseName = routingTable.database();
         this.rediscovery = rediscovery;
         this.connectionPool = connectionPool;
         this.routingTables = routingTables;
@@ -72,7 +74,7 @@ public synchronized CompletionStage<RoutingTable> freshRoutingTable( AccessMode
         else if ( routingTable.isStaleFor( mode ) )
         {
             // existing routing table is not fresh and should be updated
-            log.info( "Routing table is stale. %s", routingTable );
+            log.info( "Routing table for database '%s' is stale. %s", databaseName, routingTable );
 
             CompletableFuture<RoutingTable> resultFuture = new CompletableFuture<>();
             refreshRoutingTableFuture = resultFuture;
@@ -107,7 +109,7 @@ private synchronized void freshClusterCompositionFetched( ClusterComposition com
             routingTable.update( composition );
             connectionPool.retainAll( routingTables.allServers() );
 
-            log.info( "Updated routing table. %s", routingTable );
+            log.info( "Updated routing table for database '%s'. %s", databaseName, routingTable );
 
             CompletableFuture<RoutingTable> routingTableFuture = refreshRoutingTableFuture;
             refreshRoutingTableFuture = null;
@@ -121,13 +123,22 @@ private synchronized void freshClusterCompositionFetched( ClusterComposition com
 
     private synchronized void clusterCompositionLookupFailed( Throwable error )
     {
+        log.error( String.format( "Failed to update routing table for database '%s'. Current routing table: %s.", databaseName, routingTable ), error );
+        routingTables.remove( databaseName );
         CompletableFuture<RoutingTable> routingTableFuture = refreshRoutingTableFuture;
         refreshRoutingTableFuture = null;
         routingTableFuture.completeExceptionally( error );
     }
 
+    // This method cannot be synchronized as it will visited by all routing table's threads concurrently
     public Set<BoltServerAddress> servers()
     {
         return routingTable.servers();
     }
+
+    // for testing only
+    RoutingTable routingTable()
+    {
+        return routingTable;
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingTableHandler.java

@@ -33,6 +33,7 @@ public interface RoutingTables
 {
     /**
      * Fresh the routing table for the database and given access mode.
+     * For server version lower than 4.0, the database name will be ignored while refresh routing table.
      * @return The future of a new routing table.
      */
     CompletionStage<RoutingTable> freshRoutingTable( String databaseName, AccessMode mode );
@@ -47,4 +48,6 @@ public interface RoutingTables
      * @return the routing error handler of the given database
      */
     RoutingErrorHandler routingErrorHandler( String databaseName );
+
+    void remove( String databaseName );
 }

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingTables.java

@@ -29,15 +29,16 @@
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.internal.RoutingErrorHandler;
 import org.neo4j.driver.internal.spi.ConnectionPool;
+import org.neo4j.driver.internal.util.Clock;
 
 public class RoutingTablesImpl implements RoutingTables
 {
     private final ConcurrentMap<String,RoutingTableHandler> routingTables; //TODO ever growing map?
     private final RoutingTableHandlerFactory factory;
 
-    public RoutingTablesImpl( ConnectionPool connectionPool, RoutingTableFactory routingTableFactory, Rediscovery rediscovery, Logger log )
+    public RoutingTablesImpl( ConnectionPool connectionPool, Rediscovery rediscovery, BoltServerAddress initialRouter, Clock clock, Logger log )
     {
-        this( new ConcurrentHashMap<>(), new RoutingTableHandlerFactory( connectionPool, routingTableFactory, rediscovery, log ) );
+        this( new ConcurrentHashMap<>(), new RoutingTableHandlerFactory( connectionPool, rediscovery, initialRouter, clock, log ) );
     }
 
     RoutingTablesImpl( ConcurrentMap<String,RoutingTableHandler> routingTables, RoutingTableHandlerFactory factory )
@@ -49,13 +50,15 @@ public RoutingTablesImpl( ConnectionPool connectionPool, RoutingTableFactory rou
     @Override
     public CompletionStage<RoutingTable> freshRoutingTable( String databaseName, AccessMode mode )
     {
-        RoutingTableHandler handler = routingTableHandler( databaseName );
+        RoutingTableHandler handler = getOrCreate( databaseName );
         return handler.freshRoutingTable( mode );
     }
 
     @Override
     public Set<BoltServerAddress> allServers()
     {
+        // obviously we just had a snapshot of all servers in all routing tables
+        // after we read it, the set could already be changed.
         Set<BoltServerAddress> servers = new HashSet<>();
         for ( RoutingTableHandler tableHandler : routingTables.values() )
         {
@@ -67,32 +70,46 @@ public Set<BoltServerAddress> allServers()
     @Override
     public RoutingErrorHandler routingErrorHandler( String databaseName )
     {
-        return routingTableHandler( databaseName );
+        RoutingTableHandler handler = routingTables.get( databaseName );
+        if ( handler == null )
+        {
+            throw new IllegalStateException( String.format( "No entry for database '%s' found in routing tables.", databaseName ) );
+        }
+        return handler;
+    }
+
+    @Override
+    public void remove( String databaseName )
+    {
+        routingTables.remove( databaseName );
     }
 
-    private RoutingTableHandler routingTableHandler( String databaseName )
+    private RoutingTableHandler getOrCreate( String databaseName )
     {
         return routingTables.computeIfAbsent( databaseName, name -> factory.newInstance( name, this ) );
     }
 
     static class RoutingTableHandlerFactory
     {
         private final ConnectionPool connectionPool;
-        private final RoutingTableFactory routingTableFactory;
         private final Rediscovery rediscovery;
         private final Logger log;
+        private final BoltServerAddress initialRouter;
+        private final Clock clock;
 
-        RoutingTableHandlerFactory( ConnectionPool connectionPool, RoutingTableFactory routingTableFactory, Rediscovery rediscovery, Logger log )
+        RoutingTableHandlerFactory( ConnectionPool connectionPool, Rediscovery rediscovery, BoltServerAddress initialRouter, Clock clock, Logger log )
         {
             this.connectionPool = connectionPool;
-            this.routingTableFactory = routingTableFactory;
             this.rediscovery = rediscovery;
+            this.initialRouter = initialRouter;
+            this.clock = clock;
             this.log = log;
         }
 
         RoutingTableHandler newInstance( String databaseName, RoutingTables allTables )
         {
-            return new RoutingTableHandler( routingTableFactory.newInstance( databaseName ), rediscovery, connectionPool, allTables, log );
+            ClusterRoutingTable routingTable = new ClusterRoutingTable( databaseName, clock, initialRouter );
+            return new RoutingTableHandler( routingTable, rediscovery, connectionPool, allTables, log );
         }
     }
 }