Fix OauthValidatorCallbackHandler failure caused by apache/pulsar#19295

Author: Demogorgon314

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/security/oauth/OauthValidatorCallbackHandler.java

@@ -21,6 +21,9 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import javax.naming.AuthenticationException;
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.UnsupportedCallbackException;
@@ -126,8 +129,11 @@ protected void handleCallback(KopOAuthBearerValidatorCallback callback) {
         final String tenant = tokenAndTenant.getRight();
 
         try {
+            AuthData authData = AuthData.of(token.getBytes(StandardCharsets.UTF_8));
             final AuthenticationState authState = authenticationProvider.newAuthState(
-                    AuthData.of(token.getBytes(StandardCharsets.UTF_8)), null, null);
+                    authData, null, null);
+            // TODO: Use the configurable timeout
+            authState.authenticateAsync(authData).get(10, TimeUnit.SECONDS);
             final String role = authState.getAuthRole();
             AuthenticationDataSource authDataSource = authState.getAuthDataSource();
             callback.token(new KopOAuthBearerToken() {
@@ -168,7 +174,7 @@ public Long startTimeMs() {
                     return Long.MAX_VALUE;
                 }
             });
-        } catch (AuthenticationException e) {
+        } catch (AuthenticationException | InterruptedException | ExecutionException | TimeoutException e) {
             log.error("OAuth validator callback handler new auth state failed: ", e);
             throw new OAuthBearerIllegalTokenException(OAuthBearerValidationResult.newFailure(e.getMessage()));
         }