Abort discovery on bookmark failures and continue on authorization expired error (neo4j#1043)

This update ensures that discovery gets aborted on `ClientException` with the following codes:
- `Neo.ClientError.Transaction.InvalidBookmark`
- `Neo.ClientError.Transaction.InvalidBookmarkMixture`

In addition, it makes sure that it continues on `AuthorizationExpiredException`.

All security exceptions are mapped to `SecurityException`.

Author: injectives

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

@@ -33,6 +33,8 @@
 
 import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.Logger;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.DiscoveryException;
 import org.neo4j.driver.exceptions.FatalDiscoveryException;
 import org.neo4j.driver.exceptions.SecurityException;
@@ -59,6 +61,8 @@ public class RediscoveryImpl implements Rediscovery
     private static final String RECOVERABLE_DISCOVERY_ERROR_WITH_SERVER = "Received a recoverable discovery error with server '%s', " +
                                                                           "will continue discovery with other routing servers if available. " +
                                                                           "Complete failure is reported separately from this entry.";
+    private static final String INVALID_BOOKMARK_CODE = "Neo.ClientError.Transaction.InvalidBookmark";
+    private static final String INVALID_BOOKMARK_MIXTURE_CODE = "Neo.ClientError.Transaction.InvalidBookmarkMixture";
 
     private final BoltServerAddress initialRouter;
     private final RoutingSettings settings;
@@ -279,9 +283,8 @@ private CompletionStage<ClusterComposition> lookupOnRouter( BoltServerAddress ro
     private ClusterComposition handleRoutingProcedureError( Throwable error, RoutingTable routingTable,
                                                             BoltServerAddress routerAddress, Throwable baseError )
     {
-        if ( error instanceof SecurityException || error instanceof FatalDiscoveryException )
+        if ( mustAbortDiscovery( error ) )
         {
-            // auth error or routing error happened, terminate the discovery procedure immediately
             throw new CompletionException( error );
         }
 
@@ -295,6 +298,27 @@ private ClusterComposition handleRoutingProcedureError( Throwable error, Routing
         return null;
     }
 
+    private boolean mustAbortDiscovery( Throwable throwable )
+    {
+        boolean abort = false;
+
+        if ( !(throwable instanceof AuthorizationExpiredException) && throwable instanceof SecurityException )
+        {
+            abort = true;
+        }
+        else if ( throwable instanceof FatalDiscoveryException )
+        {
+            abort = true;
+        }
+        else if ( throwable instanceof ClientException )
+        {
+            String code = ((ClientException) throwable).code();
+            abort = INVALID_BOOKMARK_CODE.equals( code ) || INVALID_BOOKMARK_MIXTURE_CODE.equals( code );
+        }
+
+        return abort;
+    }
+
     @Override
     public List<BoltServerAddress> resolve() throws UnknownHostException
     {

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

@@ -31,6 +31,7 @@
 import org.neo4j.driver.exceptions.FatalDiscoveryException;
 import org.neo4j.driver.exceptions.Neo4jException;
 import org.neo4j.driver.exceptions.ResultConsumedException;
+import org.neo4j.driver.exceptions.SecurityException;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.exceptions.TransientException;
 
@@ -64,25 +65,34 @@ public static ResultConsumedException newResultConsumedError()
 
     public static Neo4jException newNeo4jError( String code, String message )
     {
-        String classification = extractClassification( code );
-        switch ( classification )
+        switch ( extractErrorClass( code ) )
         {
         case "ClientError":
-            if ( code.equalsIgnoreCase( "Neo.ClientError.Security.Unauthorized" ) )
+            if ( "Security".equals( extractErrorSubClass( code ) ) )
             {
-                return new AuthenticationException( code, message );
-            }
-            else if ( code.equalsIgnoreCase( "Neo.ClientError.Database.DatabaseNotFound" ) )
-            {
-                return new FatalDiscoveryException( code, message );
-            }
-            else if ( code.equalsIgnoreCase( "Neo.ClientError.Security.AuthorizationExpired" ) )
-            {
-                return new AuthorizationExpiredException( code, message );
+                if ( code.equalsIgnoreCase( "Neo.ClientError.Security.Unauthorized" ) )
+                {
+                    return new AuthenticationException( code, message );
+                }
+                else if ( code.equalsIgnoreCase( "Neo.ClientError.Security.AuthorizationExpired" ) )
+                {
+                    return new AuthorizationExpiredException( code, message );
+                }
+                else
+                {
+                    return new SecurityException( code, message );
+                }
             }
             else
             {
-                return new ClientException( code, message );
+                if ( code.equalsIgnoreCase( "Neo.ClientError.Database.DatabaseNotFound" ) )
+                {
+                    return new FatalDiscoveryException( code, message );
+                }
+                else
+                {
+                    return new ClientException( code, message );
+                }
             }
         case "TransientError":
             return new TransientException( code, message );
@@ -135,7 +145,7 @@ private static boolean isClientOrTransientError( Neo4jException error )
         return errorCode != null && (errorCode.contains( "ClientError" ) || errorCode.contains( "TransientError" ));
     }
 
-    private static String extractClassification( String code )
+    private static String extractErrorClass( String code )
     {
         String[] parts = code.split( "\\." );
         if ( parts.length < 2 )
@@ -145,6 +155,16 @@ private static String extractClassification( String code )
         return parts[1];
     }
 
+    private static String extractErrorSubClass( String code )
+    {
+        String[] parts = code.split( "\\." );
+        if ( parts.length < 3 )
+        {
+            return "";
+        }
+        return parts[2];
+    }
+
     public static void addSuppressed( Throwable mainError, Throwable error )
     {
         if ( mainError != error )

driver/src/test/java/org/neo4j/driver/internal/cluster/RediscoveryTest.java

@@ -20,6 +20,8 @@
 
 import io.netty.util.concurrent.GlobalEventExecutor;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.ArgumentCaptor;
 
 import java.io.IOException;
@@ -32,6 +34,8 @@
 
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.exceptions.AuthenticationException;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.DiscoveryException;
 import org.neo4j.driver.exceptions.ProtocolException;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
@@ -137,17 +141,59 @@ void shouldFailImmediatelyOnAuthError()
         RoutingTable table = routingTableMock( A, B, C );
 
         AuthenticationException error = assertThrows( AuthenticationException.class,
-                () -> await( rediscovery.lookupClusterComposition( table, pool, empty() ) ) );
+                                                      () -> await( rediscovery.lookupClusterComposition( table, pool, empty() ) ) );
         assertEquals( authError, error );
         verify( table ).forget( A );
     }
 
+    @Test
+    void shouldUseAnotherRouterOnAuthorizationExpiredException()
+    {
+        ClusterComposition expectedComposition =
+                new ClusterComposition( 42, asOrderedSet( A, B, C ), asOrderedSet( B, C, D ), asOrderedSet( A, B ) );
+
+        Map<BoltServerAddress,Object> responsesByAddress = new HashMap<>();
+        responsesByAddress.put( A, new AuthorizationExpiredException( "Neo.ClientError.Security.AuthorizationExpired", "message" ) );
+        responsesByAddress.put( B, expectedComposition );
+
+        ClusterCompositionProvider compositionProvider = compositionProviderMock( responsesByAddress );
+        Rediscovery rediscovery = newRediscovery( A, compositionProvider, mock( ServerAddressResolver.class ) );
+        RoutingTable table = routingTableMock( A, B, C );
+
+        ClusterComposition actualComposition = await( rediscovery.lookupClusterComposition( table, pool, empty() ) ).getClusterComposition();
+
+        assertEquals( expectedComposition, actualComposition );
+        verify( table ).forget( A );
+        verify( table, never() ).forget( B );
+        verify( table, never() ).forget( C );
+    }
+
+    @ParameterizedTest
+    @ValueSource( strings = {"Neo.ClientError.Transaction.InvalidBookmark", "Neo.ClientError.Transaction.InvalidBookmarkMixture"} )
+    void shouldFailImmediatelyOnBookmarkErrors( String code )
+    {
+        ClientException error = new ClientException( code, "Invalid" );
+
+        Map<BoltServerAddress,Object> responsesByAddress = new HashMap<>();
+        responsesByAddress.put( A, new RuntimeException( "Hi!" ) );
+        responsesByAddress.put( B, error );
+
+        ClusterCompositionProvider compositionProvider = compositionProviderMock( responsesByAddress );
+        Rediscovery rediscovery = newRediscovery( A, compositionProvider, mock( ServerAddressResolver.class ) );
+        RoutingTable table = routingTableMock( A, B, C );
+
+        ClientException actualError = assertThrows( ClientException.class,
+                                                    () -> await( rediscovery.lookupClusterComposition( table, pool, empty() ) ) );
+        assertEquals( error, actualError );
+        verify( table ).forget( A );
+    }
+
     @Test
     void shouldFallbackToInitialRouterWhenKnownRoutersFail()
     {
         BoltServerAddress initialRouter = A;
         ClusterComposition expectedComposition = new ClusterComposition( 42,
-                asOrderedSet( C, B, A ), asOrderedSet( A, B ), asOrderedSet( D, E ) );
+                                                                         asOrderedSet( C, B, A ), asOrderedSet( A, B ), asOrderedSet( D, E ) );
 
         Map<BoltServerAddress,Object> responsesByAddress = new HashMap<>();
         responsesByAddress.put( B, new ServiceUnavailableException( "Hi!" ) ); // first -> non-fatal failure

testkit-backend/src/main/java/neo4j/org/testkit/backend/messages/requests/GetFeatures.java

@@ -41,7 +41,8 @@ public class GetFeatures implements TestkitRequest
             "Temporary:DriverFetchSize",
             "Temporary:DriverMaxTxRetryTime",
             "Feature:Auth:Kerberos",
-            "Feature:Auth:Custom"
+            "Feature:Auth:Custom",
+            "Temporary:FastFailingDiscovery"
     ) );
 
     private static final Set<String> SYNC_FEATURES = new HashSet<>( Arrays.asList(