Add onAuthenticationFailure callback to Authenticator

core/src/main/java/com/datastax/oss/driver/api/core/auth/Authenticator.java

@@ -39,6 +39,9 @@
  *   <li>When the server indicates that authentication is successful, the {@link
  *       #onAuthenticationSuccess} method will be called with the last information that the server
  *       may optionally have sent.
+ *   <li>When the server indicates that authentication has failed, the {@link
+ *       #onAuthenticationFailure} method will be called with {@link AuthenticationException}
+ *       encapsulating the server error message
  * </ol>
  *
  * The exact nature of the negotiation between client and server is specific to the authentication
@@ -93,4 +96,11 @@ public interface Authenticator {
    */
   @NonNull
   CompletionStage<Void> onAuthenticationSuccess(@Nullable ByteBuffer token);
+
+  /**
+   * Called when authentication fails.
+   *
+   * @param exception contains authentication error message
+   */
+  default void onAuthenticationFailure(AuthenticationException exception) {}
 }

core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java

@@ -280,12 +280,14 @@ void onResponse(Message response) {
         } else if (step == Step.AUTH_RESPONSE
             && response instanceof Error
             && ((Error) response).code == ProtocolConstants.ErrorCode.AUTH_ERROR) {
-          fail(
+          AuthenticationException cause =
               new AuthenticationException(
                   endPoint,
                   String.format(
                       "server replied with '%s' to AuthResponse request",
-                      ((Error) response).message)));
+                      ((Error) response).message));
+          authenticator.onAuthenticationFailure(cause);
+          fail(cause);
         } else if (step == Step.GET_CLUSTER_NAME && response instanceof Rows) {
           Rows rows = (Rows) response;
           List<ByteBuffer> row = Objects.requireNonNull(rows.getData().poll());

core/src/test/java/com/datastax/oss/driver/internal/core/channel/MockAuthenticator.java

@@ -15,6 +15,7 @@
  */
 package com.datastax.oss.driver.internal.core.channel;
 
+import com.datastax.oss.driver.api.core.auth.AuthenticationException;
 import com.datastax.oss.driver.api.core.auth.SyncAuthenticator;
 import com.datastax.oss.protocol.internal.util.Bytes;
 import java.nio.ByteBuffer;
@@ -30,6 +31,7 @@ public class MockAuthenticator implements SyncAuthenticator {
   static final String INITIAL_RESPONSE = "0xcafebabe";
 
   volatile String successToken;
+  volatile boolean authFailure;
 
   @Override
   public ByteBuffer initialResponseSync() {
@@ -45,4 +47,9 @@ public ByteBuffer evaluateChallengeSync(ByteBuffer challenge) {
   public void onAuthenticationSuccessSync(ByteBuffer token) {
     successToken = Bytes.toHexString(token);
   }
+
+  @Override
+  public void onAuthenticationFailure(AuthenticationException exception) {
+    authFailure = true;
+  }
 }

core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java

@@ -409,6 +409,49 @@ public void should_fail_to_initialize_if_server_sends_auth_error() throws Throwa
                             END_POINT)));
   }
 
+  @Test
+  public void should_invoke_authenticator_if_server_sends_auth_error() throws Throwable {
+    channel
+        .pipeline()
+        .addLast(
+            ChannelFactory.INIT_HANDLER_NAME,
+            new ProtocolInitHandler(
+                internalDriverContext,
+                DefaultProtocolVersion.V4,
+                null,
+                END_POINT,
+                DriverChannelOptions.DEFAULT,
+                heartbeatHandler,
+                false));
+
+    String serverAuthenticator = "mockServerAuthenticator";
+    AuthProvider authProvider = mock(AuthProvider.class);
+    MockAuthenticator authenticator = new MockAuthenticator();
+    when(authProvider.newAuthenticator(END_POINT, serverAuthenticator)).thenReturn(authenticator);
+    when(internalDriverContext.getAuthProvider()).thenReturn(Optional.of(authProvider));
+
+    ChannelFuture connectFuture = channel.connect(new InetSocketAddress("localhost", 9042));
+
+    Frame requestFrame = readOutboundFrame();
+    assertThat(requestFrame.message).isInstanceOf(Startup.class);
+    assertThat(connectFuture).isNotDone();
+
+    writeInboundFrame(requestFrame, new Authenticate("mockServerAuthenticator"));
+
+    requestFrame = readOutboundFrame();
+    assertThat(requestFrame.message).isInstanceOf(AuthResponse.class);
+    assertThat(connectFuture).isNotDone();
+
+    writeInboundFrame(
+        requestFrame, new Error(ProtocolConstants.ErrorCode.AUTH_ERROR, "mock error"));
+
+    assertThat(connectFuture)
+        .isFailed(e -> assertThat(e).isInstanceOf(AuthenticationException.class));
+
+    // verify that onAuthenticationFailure callback was invoked
+    assertThat(authenticator.authFailure).isTrue();
+  }
+
   @Test
   public void should_check_cluster_name_if_provided() {
     channel