Adding more boltkit tests for multi-databases

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/exceptions/RoutingException.java

@@ -23,15 +23,15 @@
  * This exception should not be retried.
  * @since 2.0
  */
-public class RoutingException extends Neo4jException
+public class RoutingException extends ClientException
 {
     public RoutingException( String message )
     {
         super( message );
     }
 
-    public RoutingException( String message, Throwable throwable )
+    public RoutingException( String code, String message )
     {
-        super( message, throwable);
+        super( code, message );
     }
 }

driver/src/main/java/org/neo4j/driver/exceptions/SecurityException.java

@@ -24,7 +24,7 @@
  * Restart of server/driver/cluster might be required to recover from this error.
  * @since 1.1
  */
-public class SecurityException extends Neo4jException
+public class SecurityException extends ClientException
 {
     public SecurityException( String code, String message )
     {

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

@@ -24,5 +24,5 @@
 
 public interface ClusterCompositionProvider
 {
-    CompletionStage<ClusterCompositionResponse> getClusterComposition( CompletionStage<Connection> connectionStage, String databaseName );
+    CompletionStage<ClusterComposition> getClusterComposition( CompletionStage<Connection> connectionStage, String databaseName );
 }

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

@@ -70,7 +70,12 @@ public boolean isStaleFor( AccessMode mode )
     @Override
     public boolean isStale( long staleRoutingTableTimeout )
     {
-        return expirationTimeout + staleRoutingTableTimeout < clock.millis();
+        long expireTime = expirationTimeout + staleRoutingTableTimeout;
+        if ( expireTime < 0 )
+        {
+            expireTime = Long.MAX_VALUE;
+        }
+        return  expireTime < clock.millis();
     }
 
     @Override
@@ -133,7 +138,7 @@ public void removeWriter( BoltServerAddress toRemove )
     @Override
     public synchronized String toString()
     {
-        return format( "Ttl %s, currentTime %s, routers %s, writers %s, readers %s, database '%s'.",
+        return format( "Ttl %s, currentTime %s, routers %s, writers %s, readers %s, database '%s'",
                 expirationTimeout, clock.millis(), routers, writers, readers, databaseName );
     }
 }

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

@@ -31,6 +31,7 @@
 import java.util.stream.Stream;
 
 import org.neo4j.driver.Logger;
+import org.neo4j.driver.exceptions.RoutingException;
 import org.neo4j.driver.exceptions.SecurityException;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.internal.BoltServerAddress;
@@ -43,13 +44,16 @@
 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;
 
+/**
+ * This class is used by all router tables to perform discovery.
+ * In other words, the methods in this class could be called by multiple threads concurrently.
+ */
 public class RediscoveryImpl implements Rediscovery
 {
-    private static final String NO_ROUTERS_AVAILABLE = "Could not perform discovery. No routing servers available.";
+    private static final String NO_ROUTERS_AVAILABLE = "Could not perform discovery for database '%s'. No routing servers available.";
 
     private final BoltServerAddress initialRouter;
     private final RoutingSettings settings;
@@ -58,7 +62,7 @@ public class RediscoveryImpl implements Rediscovery
     private final ServerAddressResolver resolver;
     private final EventExecutorGroup eventExecutorGroup;
 
-    private volatile boolean useInitialRouter;
+    private volatile boolean useInitialRouter; // TODO thread safe?
 
     public RediscoveryImpl( BoltServerAddress initialRouter, RoutingSettings settings, ClusterCompositionProvider provider,
             EventExecutorGroup eventExecutorGroup, ServerAddressResolver resolver, Logger logger )
@@ -114,7 +118,7 @@ else if ( composition != null )
                 int newFailures = failures + 1;
                 if ( newFailures >= settings.maxRoutingFailures() )
                 {
-                    result.completeExceptionally( new ServiceUnavailableException( NO_ROUTERS_AVAILABLE ) );
+                    result.completeExceptionally( new ServiceUnavailableException( String.format( NO_ROUTERS_AVAILABLE, routingTable.database() ) ) );
                 }
                 else
                 {
@@ -247,25 +251,24 @@ private CompletionStage<ClusterComposition> lookupOnRouter( BoltServerAddress ro
             }
             else
             {
-                return response.clusterComposition();
+                return response;
             }
         } );
     }
 
     private ClusterComposition handleRoutingProcedureError( Throwable error, RoutingTable routingTable,
             BoltServerAddress routerAddress )
     {
-        if ( error instanceof SecurityException || isRoutingError( error ) )
+        if ( error instanceof SecurityException || error instanceof RoutingException )
         {
             // auth error or routing error happened, terminate the discovery procedure immediately
             throw new CompletionException( error );
         }
-        {
-            // connection turned out to be broken
-            logger.info( format( "Failed to connect to routing server '%s'.", routerAddress ), error );
-            routingTable.forget( routerAddress );
-            return null;
-        }
+
+        // Retriable error happened during discovery.
+        logger.warn( format( "Failed to update routing table with server '%s'.", routerAddress ), error );
+        routingTable.forget( routerAddress );
+        return null;
     }
 
     private List<BoltServerAddress> resolve( BoltServerAddress address )

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

@@ -19,15 +19,15 @@
 package org.neo4j.driver.internal.cluster;
 
 import java.util.List;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 
-import org.neo4j.driver.internal.spi.Connection;
-import org.neo4j.driver.internal.util.Clock;
 import org.neo4j.driver.Record;
 import org.neo4j.driver.Statement;
 import org.neo4j.driver.exceptions.ProtocolException;
-import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.exceptions.value.ValueException;
+import org.neo4j.driver.internal.spi.Connection;
+import org.neo4j.driver.internal.util.Clock;
 
 import static java.lang.String.format;
 
@@ -50,21 +50,19 @@ public RoutingProcedureClusterCompositionProvider( Clock clock, RoutingSettings
     }
 
     @Override
-    public CompletionStage<ClusterCompositionResponse> getClusterComposition( CompletionStage<Connection> connectionStage, String databaseName )
+    public CompletionStage<ClusterComposition> getClusterComposition( CompletionStage<Connection> connectionStage, String databaseName )
     {
         return routingProcedureRunner.run( connectionStage, databaseName )
                 .thenApply( this::processRoutingResponse );
     }
 
-    private ClusterCompositionResponse processRoutingResponse( RoutingProcedureResponse response )
+    private ClusterComposition processRoutingResponse( RoutingProcedureResponse response )
     {
         if ( !response.isSuccess() )
         {
-            return new ClusterCompositionResponse.Failure( new ServiceUnavailableException( format(
-                    "Failed to run '%s' on server. " +
-                    "Please make sure that there is a Neo4j server or cluster up running.",
-                    invokedProcedureString( response ) ), response.error()
-            ) );
+            throw new CompletionException( format(
+                    "Failed to run '%s' on server. Please make sure that there is a Neo4j server or cluster up running.",
+                    invokedProcedureString( response ) ), response.error() );
         }
 
         List<Record> records = response.records();
@@ -74,9 +72,9 @@ private ClusterCompositionResponse processRoutingResponse( RoutingProcedureRespo
         // the record size is wrong
         if ( records.size() != 1 )
         {
-            return new ClusterCompositionResponse.Failure( new ProtocolException( format(
+            throw new ProtocolException( format(
                     PROTOCOL_ERROR_MESSAGE + "records received '%s' is too few or too many.",
-                    invokedProcedureString( response ), records.size() ) ) );
+                    invokedProcedureString( response ), records.size() ) );
         }
 
         // failed to parse the record
@@ -87,21 +85,21 @@ private ClusterCompositionResponse processRoutingResponse( RoutingProcedureRespo
         }
         catch ( ValueException e )
         {
-            return new ClusterCompositionResponse.Failure( new ProtocolException( format(
+            throw new ProtocolException( format(
                     PROTOCOL_ERROR_MESSAGE + "unparsable record received.",
-                    invokedProcedureString( response ) ), e ) );
+                    invokedProcedureString( response ) ), e );
         }
 
         // the cluster result is not a legal reply
         if ( !cluster.hasRoutersAndReaders() )
         {
-            return new ClusterCompositionResponse.Failure( new ProtocolException( format(
+            throw new ProtocolException( format(
                     PROTOCOL_ERROR_MESSAGE + "no router or reader found in response.",
-                    invokedProcedureString( response ) ) ) );
+                    invokedProcedureString( response ) ) );
         }
 
         // all good
-        return new ClusterCompositionResponse.Success( cluster );
+        return cluster;
     }
 
     private static String invokedProcedureString( RoutingProcedureResponse response )

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

@@ -43,6 +43,7 @@ public class RoutingTableHandler implements RoutingErrorHandler
     private final Logger log;
 
     // This defines how long we shall wait before trimming a routing table from routing tables after it is stale.
+    // TODO make this a configuration option
     public static final Duration STALE_ROUTING_TABLE_PURGE_TIMEOUT = Duration.ofSeconds( 30 );
 
 

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

@@ -24,8 +24,8 @@
 
 public final class MultiDatabaseUtil
 {
-    public static final String ABSENT_DB_NAME = "";
-    public static final String SYSTEM_DB_NAME = "system";
+    public static final String ABSENT_DB_NAME = ""; // TODO _default
+    public static final String SYSTEM_DB_NAME = "system"; // TODO _dbms
 
     public static void assertEmptyDatabaseName( String databaseName, int boltVersion )
     {

driver/src/main/java/org/neo4j/driver/internal/messaging/request/MultiDatabaseUtil.java

@@ -63,6 +63,10 @@ public static Neo4jException newNeo4jError( String code, String message )
             {
                 return new AuthenticationException( code, message );
             }
+            else if ( code.equalsIgnoreCase( "Neo.ClientError.Database.DatabaseNotFound" ) )
+            {
+                return new RoutingException( code, message );
+            }
             else
             {
                 return new ClientException( code, message );
@@ -91,23 +95,6 @@ public static boolean isFatal( Throwable error )
         return true;
     }
 
-    public static boolean isRoutingError( Throwable error )
-    {
-        if ( error instanceof RoutingException  )
-        {
-            return true;
-        }
-        else if ( error instanceof Neo4jException )
-        {
-            String errorCode = ((Neo4jException) error).code();
-            return errorCode != null && (errorCode.startsWith( "Neo.ClientError.Database.DatabaseNotFound" ));
-        }
-        else
-        {
-            return false;
-        }
-    }
-
     public static void rethrowAsyncException( ExecutionException e )
     {
         Throwable error = e.getCause();