Abort discovery on bookmark failures and continue on authorization expired error

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`.

Author: injectives

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

@@ -34,6 +34,8 @@
 import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
+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;
@@ -61,6 +63,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;
@@ -278,10 +282,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 ||
-             (error instanceof IllegalStateException && ConnectionPool.CONNECTION_POOL_CLOSED_ERROR_MESSAGE.equals( error.getMessage() )) )
+        if ( mustAbortDiscovery( error ) )
         {
-            // auth error or routing error happened, terminate the discovery procedure immediately
             throw new CompletionException( error );
         }
 
@@ -295,6 +297,31 @@ 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 IllegalStateException && ConnectionPool.CONNECTION_POOL_CLOSED_ERROR_MESSAGE.equals( throwable.getMessage() ) )
+        {
+            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/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;
@@ -33,6 +35,8 @@
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
 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;
@@ -143,6 +147,67 @@ void shouldFailImmediatelyOnAuthError()
         verify( table ).forget( A );
     }
 
+    @Test
+    void shouldUseAnotherRouterOnAuthorizationExpiredException()
+    {
+        ClusterComposition expectedComposition =
+                new ClusterComposition( 42, asOrderedSet( A, B, C ), asOrderedSet( B, C, D ), asOrderedSet( A, B ), null );
+
+        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(), null ) ).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(), null ) ) );
+        assertEquals( error, actualError );
+        verify( table ).forget( A );
+    }
+
+    @Test
+    void shouldFailImmediatelyOnClosedPoolError()
+    {
+        IllegalStateException error = new IllegalStateException( ConnectionPool.CONNECTION_POOL_CLOSED_ERROR_MESSAGE );
+
+        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 );
+
+        IllegalStateException actualError = assertThrows( IllegalStateException.class,
+                                                          () -> await( rediscovery.lookupClusterComposition( table, pool, empty(), null ) ) );
+        assertEquals( error, actualError );
+        verify( table ).forget( A );
+    }
+
     @Test
     void shouldFallbackToInitialRouterWhenKnownRoutersFail()
     {