fix: changed/fixed servicename for createSaslServer, createSaslClient

Author: SML0127

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/jaas/DefaultLogin.java

@@ -25,9 +25,12 @@
 import javax.annotation.Nullable;
 import javax.security.auth.Subject;
 import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.kerberos.KerberosPrincipal;
 import javax.security.auth.login.LoginContext;
 import javax.security.auth.login.LoginException;
 
+import java.util.Set;
+
 /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
  * additional information regarding copyright ownership. */
@@ -73,9 +76,26 @@ public Subject subject() {
 
     @Override
     public String serviceName() {
+        if (loginContext != null && loginContext.getSubject() != null) {
+            Set<KerberosPrincipal> principals =
+                    loginContext.getSubject().getPrincipals(KerberosPrincipal.class);
+            if (!principals.isEmpty()) {
+                KerberosPrincipal principal = principals.iterator().next();
+                String name = principal.getName();
+                int slash = name.indexOf('/');
+                if (slash > 0) {
+                    return name.substring(0, slash);
+                }
+                int at = name.indexOf('@');
+                if (at > 0) {
+                    return name.substring(0, at);
+                }
+                return name;
+            }
+        }
         return contextName;
     }
 
     @Override
     public void close() {}
-}
+}

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/jaas/SaslServerFactory.java

@@ -19,7 +19,6 @@
 
 import org.apache.fluss.security.auth.sasl.gssapi.GssapiServerCallbackHandler;
 import org.apache.fluss.security.auth.sasl.plain.PlainServerCallbackHandler;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,7 +28,6 @@
 import javax.security.sasl.SaslClient;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
-
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
 import java.util.Arrays;
@@ -71,7 +69,7 @@ public static SaslServer createSaslServer(
                                     () ->
                                             Sasl.createSaslServer(
                                                     mechanism,
-                                                    "fluss",
+                                                    loginManager.serviceName(),
                                                     hostName,
                                                     props,
                                                     callbackHandler));
@@ -99,6 +97,8 @@ public static SaslClient createSaslClient(
                 (PrivilegedExceptionAction<SaslClient>)
                         () -> {
                             String[] mechs = {mechanism};
+                            // The serviceName here is the name of the service we are connecting to.
+                            // It is NOT the name of the client principal.
                             String serviceName = "fluss";
                             LOG.debug(
                                     "Creating SaslClient: service={};mechs={}",