MINOR: add Utils.isBlank to check whitespace character string and empty string (apache#10012)

Reviewers: Chia-Ping Tsai <[email protected]>

Author: tang7526

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredValidatorCallbackHandler.java

@@ -30,6 +30,7 @@
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallback;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -171,32 +172,23 @@ private void handleCallback(OAuthBearerValidatorCallback callback) {
         OAuthBearerValidationUtils.validateTimeConsistency(unsecuredJwt).throwExceptionIfFailed();
         OAuthBearerValidationUtils.validateScope(unsecuredJwt, requiredScope).throwExceptionIfFailed();
         log.info("Successfully validated token with principal {}: {}", unsecuredJwt.principalName(),
-                unsecuredJwt.claims().toString());
+                unsecuredJwt.claims());
         callback.token(unsecuredJwt);
     }
 
     private String principalClaimName() {
         String principalClaimNameValue = option(PRINCIPAL_CLAIM_NAME_OPTION);
-        String principalClaimName = principalClaimNameValue != null && !principalClaimNameValue.trim().isEmpty()
-                ? principalClaimNameValue.trim()
-                : "sub";
-        return principalClaimName;
+        return Utils.isBlank(principalClaimNameValue) ? "sub" : principalClaimNameValue.trim();
     }
 
     private String scopeClaimName() {
         String scopeClaimNameValue = option(SCOPE_CLAIM_NAME_OPTION);
-        String scopeClaimName = scopeClaimNameValue != null && !scopeClaimNameValue.trim().isEmpty()
-                ? scopeClaimNameValue.trim()
-                : "scope";
-        return scopeClaimName;
+        return Utils.isBlank(scopeClaimNameValue) ? "scope" : scopeClaimNameValue.trim();
     }
 
     private List<String> requiredScope() {
         String requiredSpaceDelimitedScope = option(REQUIRED_SCOPE_OPTION);
-        List<String> requiredScope = requiredSpaceDelimitedScope == null || requiredSpaceDelimitedScope.trim().isEmpty()
-            ? Collections.emptyList()
-            : OAuthBearerScopeUtils.parseScope(requiredSpaceDelimitedScope.trim());
-        return requiredScope;
+        return Utils.isBlank(requiredSpaceDelimitedScope) ? Collections.emptyList() : OAuthBearerScopeUtils.parseScope(requiredSpaceDelimitedScope.trim());
     }
 
     private int allowableClockSkewMs() {

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -1334,4 +1334,13 @@ public static long getDateTime(String timestamp) throws ParseException, IllegalA
     public static <S> Iterator<S> covariantCast(Iterator<? extends S> iterator) {
         return (Iterator<S>) iterator;
     }
+
+    /**
+     * Checks if a string is null, empty or whitespace only.
+     * @param str a string to be checked
+     * @return true if the string is null, empty or whitespace only; otherwise, return false.
+     */    
+    public static boolean isBlank(String str) {
+        return str == null || str.trim().isEmpty();
+    }
 }

clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java

@@ -851,4 +851,12 @@ private void invokeGetDateTimeMethod(final SimpleDateFormat format) throws Parse
         Utils.getDateTime(formattedCheckpoint);
     }
 
+    @Test
+    void testIsBlank() {
+        assertTrue(Utils.isBlank(null));
+        assertTrue(Utils.isBlank(""));
+        assertTrue(Utils.isBlank(" "));
+        assertFalse(Utils.isBlank("bob"));
+        assertFalse(Utils.isBlank(" bob "));
+    }
 }