[PIP 97] Update Authentication Interfaces to Include Async Authentication Methods (apache#12104)

Master Issue: apache#12105

### Motivation

As the first part of PIP-97, we need to update the interfaces. This PR is the only PR that will update interfaces. It should not introduce any breaking changes.

### Modifications

#### AuthenticationProvider

* Add `AuthenticationProvider#authenticateAsync`. Include a default implementation that calls the `authenticate` method. Note that current implementations should all be non-blocking, so there is no need to push the execution to a separate thread.
* Deprecate `AuthenticationProvider#authenticate`.
* Add `AuthenticationProvider#authenticateHttpRequestAsync`. This method is complicated. It is only called when using the SASL authentication provider (this is hard coded into the Pulsar code base). As such, I would argue that it is worth removing support for this unreachable method and then refactor the SASL authentication provider. I annotated this method with `@InterfaceStability.Unstable` and added details to the Javadoc in order to communicate the uncertainty of this method's future. I am happy to discuss this in more detail though.
* Deprecate `AuthenticationProvider#authenticateHttpRequest`.

#### AuthenticationState

* Add `AuthenticationState#authenticateAsync`. Include a default implementation that calls the `authenticate` method and then performs a check to determine what result to return. Note that current implementations should all be non-blocking, so there is no need to push the execution to a separate thread.
* Deprecate `AuthenticationState#authenticate`. The preferred method is `AuthenticationState#authenticateAsync`.
* Deprecate `AuthenticationState#isComplete`. This method can be avoided by inferring authentication completeness from the result of `AuthenticationState#authenticateAsync`. When the result is `null`, auth is complete. When it is not `null`, auth is not complete. Since the result of the `authenticateAsync` method is the body delivered to the client, this seems like a reasonable abstraction to make. As a consequence, the `AuthenticationState` is simpler and also avoids certain thread safety issues that might arise when calling `isComplete` from a different thread.

#### AuthenticationDataSource
* Deprecate `AuthenticationDataSource#authenticate`. This method is not called by the Pulsar authentication framework. This needs to be deprecated to prevent confusion for end users seeking to extend the authentication framework. There is no need for an async version of this method.

### Verifying this change

These changes only affect the interfaces. I will need to add tests to verify the correctness of the default implementations in this PR.

### Does this pull request potentially affect one of the following parts:

Yes, it affects the public API. That is why it has a PIP.

### Documentation
I've updated the Javadocs. There is not any current documentation on implementing your own authentication provider, so I think updating Javadocs is sufficient documentation, for now.

Author: michaeljmarshall

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationDataSource.java

@@ -102,7 +102,10 @@ default String getCommandData() {
     /**
      * Evaluate and challenge the data that passed in, and return processed data back.
      * It is used for mutual authentication like SASL.
+     * NOTE: this method is not called by the Pulsar authentication framework.
+     * @deprecated use {@link AuthenticationProvider} or {@link AuthenticationState}.
      */
+    @Deprecated
     default AuthData authenticate(AuthData data) throws AuthenticationException {
         throw new AuthenticationException("Not supported");
     }

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 
 import java.net.SocketAddress;
+import java.util.concurrent.CompletableFuture;
 import javax.naming.AuthenticationException;
 
 import javax.net.ssl.SSLSession;
@@ -30,6 +31,8 @@
 
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.common.api.AuthData;
+import org.apache.pulsar.common.classification.InterfaceStability;
+import org.apache.pulsar.common.util.FutureUtil;
 
 /**
  * Provider of authentication mechanism
@@ -51,6 +54,29 @@ public interface AuthenticationProvider extends Closeable {
      */
     String getAuthMethodName();
 
+    /**
+     * Validate the authentication for the given credentials with the specified authentication data.
+     * This method is useful in one stage authentication, if you're not doing one stage or if you're providing
+     * your own state implementation for one stage authentication, it should return a failed future.
+     *
+     * <p>Warning: the calling thread is an IO thread. Any implementation that relies on blocking behavior
+     * must ensure that the execution is completed using a separate thread pool to ensure IO threads
+     * are never blocked.</p>
+     *
+     * @param authData authentication data generated while initiating a connection. There are several types,
+     *                 including, but not strictly limited to, {@link AuthenticationDataHttp},
+     *                 {@link AuthenticationDataHttps}, and {@link AuthenticationDataCommand}.
+     * @return A completed future with the "role" string for the authenticated connection, if authentication is
+     * successful, or a failed future if the authData is not valid.
+     */
+    default CompletableFuture<String> authenticateAsync(AuthenticationDataSource authData) {
+        try {
+            return CompletableFuture.completedFuture(this.authenticate(authData));
+        } catch (AuthenticationException e) {
+            return FutureUtil.failedFuture(e);
+        }
+    }
+
     /**
      * Validate the authentication for the given credentials with the specified authentication data.
      * This method is useful in one stage authn, if you're not doing one stage or if you're providing
@@ -61,7 +87,9 @@ public interface AuthenticationProvider extends Closeable {
      * @return the "role" string for the authenticated connection, if the authentication was successful
      * @throws AuthenticationException
      *             if the credentials are not valid
+     * @deprecated use and implement {@link AuthenticationProvider#authenticateAsync(AuthenticationDataSource)} instead.
      */
+    @Deprecated
     default String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
         throw new AuthenticationException("Not supported");
     }
@@ -76,10 +104,38 @@ default AuthenticationState newAuthState(AuthData authData,
         return new OneStageAuthenticationState(authData, remoteAddress, sslSession, this);
     }
 
+    /**
+     * Validate the authentication for the given credentials with the specified authentication data.
+     *
+     * <p>Warning: the calling thread is an IO thread. Any implementations that rely on blocking behavior
+     * must ensure that the execution is completed on using a separate thread pool to ensure IO threads
+     * are never blocked.</p>
+     *
+     * <p>Note: this method is marked as unstable because the Pulsar code base only calls it for the
+     * Pulsar Broker Auth SASL plugin. All non SASL HTTP requests are authenticated using the
+     * {@link AuthenticationProvider#authenticateAsync(AuthenticationDataSource)} method. As such,
+     * this method might be removed in favor of the SASL provider implementing the
+     * {@link AuthenticationProvider#authenticateAsync(AuthenticationDataSource)} method.</p>
+     *
+     * @return Set response, according to passed in request.
+     * and return whether we should do following chain.doFilter or not.
+     */
+    @InterfaceStability.Unstable
+    default CompletableFuture<Boolean> authenticateHttpRequestAsync(HttpServletRequest request,
+                                                                    HttpServletResponse response) {
+        try {
+            return CompletableFuture.completedFuture(this.authenticateHttpRequest(request, response));
+        } catch (Exception e) {
+            return FutureUtil.failedFuture(e);
+        }
+    }
+
     /**
      * Set response, according to passed in request.
      * and return whether we should do following chain.doFilter or not.
+     * @deprecated use and implement {@link AuthenticationProvider#authenticateHttpRequestAsync} instead.
      */
+    @Deprecated
     default boolean authenticateHttpRequest(HttpServletRequest request, HttpServletResponse response) throws Exception {
         throw new AuthenticationException("Not supported");
     }

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationState.java

@@ -22,6 +22,9 @@
 import javax.naming.AuthenticationException;
 
 import org.apache.pulsar.common.api.AuthData;
+import org.apache.pulsar.common.util.FutureUtil;
+
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Interface for authentication state.
@@ -39,17 +42,50 @@ public interface AuthenticationState {
 
     /**
      * Challenge passed in auth data and get response data.
+     * @deprecated use and implement {@link AuthenticationState#authenticateAsync(AuthData)} instead.
      */
+    @Deprecated
     AuthData authenticate(AuthData authData) throws AuthenticationException;
 
+    /**
+     * Challenge passed in auth data. If authentication is complete after the execution of this method, return null.
+     * Otherwise, return response data to be sent to the client.
+     *
+     * <p>Note: the implementation of {@link AuthenticationState#authenticate(AuthData)} converted a null result into a
+     * zero length byte array when {@link AuthenticationState#isComplete()} returned false after authentication. In
+     * order to simplify this interface, the determination of whether to send a challenge back to the client is only
+     * based on the result of this method. In order to maintain backwards compatibility, the default implementation of
+     * this method calls {@link AuthenticationState#isComplete()} and returns a result compliant with the new
+     * paradigm.</p>
+     */
+    default CompletableFuture<AuthData> authenticateAsync(AuthData authData) {
+        try {
+            AuthData result = this.authenticate(authData);
+            if (isComplete()) {
+                return CompletableFuture.completedFuture(null);
+            } else {
+                return result != null
+                        ? CompletableFuture.completedFuture(result)
+                        : CompletableFuture.completedFuture(AuthData.of(new byte[0]));
+            }
+        } catch (Exception e) {
+            return FutureUtil.failedFuture(e);
+        }
+    }
+
     /**
      * Return AuthenticationDataSource.
      */
     AuthenticationDataSource getAuthDataSource();
 
     /**
      * Whether the authentication is completed or not.
+     * @deprecated this method's logic is captured by the result of
+     * {@link AuthenticationState#authenticateAsync(AuthData)}. When the result is a {@link CompletableFuture} with a
+     * null result, authentication is complete. When the result is a {@link CompletableFuture} with a nonnull result,
+     * authentication is incomplete and requires an auth challenge.
      */
+    @Deprecated
     boolean isComplete();
 
     /**