[CELEBORN-1318][FOLLOWUP] Authenticate bearer token directly

### What changes were proposed in this pull request?
I am working on the bearer token authentication integration, and meet the token base64 decode issue.

And found that, for bear token, we shall authenticate it directly.

![image](https://github.com/user-attachments/assets/0270f924-1d57-4ddd-9fdc-632711782078)

### Why are the changes needed?
For bearer authentication issue.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?

Integration testing.

<img width="1727" alt="image" src="https://github.com/user-attachments/assets/0c03b73b-be08-45b0-81c4-006eebc5ac3b">

Closes #2666 from turboFei/bear_auth.

Authored-by: Wang, Fei <fwang12@ebay.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

service/src/main/scala/org/apache/celeborn/server/common/http/authentication/BearerAuthenticationHandler.scala

@@ -17,10 +17,10 @@
 
 package org.apache.celeborn.server.common.http.authentication
 
-import java.nio.charset.StandardCharsets
-import java.util.Base64
 import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
 
+import org.apache.commons.lang3.StringUtils
+
 import org.apache.celeborn.common.CelebornConf
 import org.apache.celeborn.common.authentication.{AnonymousAuthenticationProviderImpl, DefaultTokenCredential}
 import org.apache.celeborn.common.authentication.HttpAuthSchemes._
@@ -71,17 +71,13 @@ class BearerAuthenticationHandler(providerClass: String)
       request: HttpServletRequest,
       response: HttpServletResponse): String = {
     var principal: String = null
-    val inputToken = Option(getAuthorization(request))
-      .map(a => Base64.getDecoder.decode(a.getBytes()))
-      .getOrElse(Array.empty[Byte])
+    val inputToken = getAuthorization(request)
 
-    if (!allowAnonymous && inputToken.isEmpty) {
+    if (!allowAnonymous && StringUtils.isBlank(inputToken)) {
       response.setHeader(WWW_AUTHENTICATE_HEADER, authScheme.toString)
       response.setStatus(HttpServletResponse.SC_UNAUTHORIZED)
     } else {
-      val credential = DefaultTokenCredential(
-        new String(inputToken, StandardCharsets.UTF_8),
-        HttpAuthUtils.getCredentialExtraInfo)
+      val credential = DefaultTokenCredential(inputToken, HttpAuthUtils.getCredentialExtraInfo)
       principal = HttpAuthenticationFactory
         .getTokenAuthenticationProvider(providerClass, conf)
         .authenticate(credential).getName

service/src/test/scala/org/apache/celeborn/server/common/http/ApiBaseResourceAuthenticationSuite.scala

@@ -57,10 +57,7 @@ abstract class ApiBaseResourceAuthenticationSuite extends HttpTestHelper {
       Base64.getEncoder.encode(s"$user:$password".getBytes()),
       StandardCharsets.UTF_8)
 
-  def bearerAuthorizationHeader(token: String): String =
-    HttpAuthSchemes.BEARER + " " + new String(
-      Base64.getEncoder.encode(token.getBytes()),
-      StandardCharsets.UTF_8)
+  def bearerAuthorizationHeader(token: String): String = HttpAuthSchemes.BEARER + " " + token
 
   Seq("conf", "listDynamicConfigs", "workerInfo", "shuffle", "applications").foreach { api =>
     test(s"API $api authentication") {