feat: received service name. default is fluss.

Author: SML0127

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1234,6 +1234,14 @@ public class ConfigOptions {
                                     + "This is used when the client connects to the Fluss cluster with SASL authentication enabled. "
                                     + "If not provided, the password will be read from the JAAS configuration string specified by `client.security.sasl.jaas.config`.");
 
+    public static final ConfigOption<String> CLIENT_KERBEROS_SERVICE_NAME =
+            key("client.security.kerberos.service.name")
+                    .stringType()
+                    .defaultValue("fluss")
+                    .withDescription(
+                            "The Kerberos principal name that the server runs as. This can be defined either in "
+                                    + "Fluss's JAAS config or in Fluss's config.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Fluss Table
     // ------------------------------------------------------------------------

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslClientAuthenticator.java

@@ -32,6 +32,7 @@
 import java.security.PrivilegedExceptionAction;
 import java.util.Map;
 
+import static org.apache.fluss.config.ConfigOptions.CLIENT_KERBEROS_SERVICE_NAME;
 import static org.apache.fluss.config.ConfigOptions.CLIENT_SASL_JAAS_CONFIG;
 import static org.apache.fluss.config.ConfigOptions.CLIENT_SASL_JAAS_PASSWORD;
 import static org.apache.fluss.config.ConfigOptions.CLIENT_SASL_JAAS_USERNAME;
@@ -45,12 +46,14 @@ public class SaslClientAuthenticator implements ClientAuthenticator {
     private final String mechanism;
     private final Map<String, String> pros;
     private final String jaasConfig;
+    private final String serviceName;
 
     private SaslClient saslClient;
     private LoginManager loginManager;
 
     public SaslClientAuthenticator(Configuration configuration) {
         this.mechanism = configuration.get(CLIENT_SASL_MECHANISM).toUpperCase();
+        this.serviceName = configuration.get(CLIENT_KERBEROS_SERVICE_NAME);
         String jaasConfigStr = configuration.getString(CLIENT_SASL_JAAS_CONFIG);
         if (jaasConfigStr == null && mechanism.equals(PlainSaslServer.PLAIN_MECHANISM)) {
             String username = configuration.get(CLIENT_SASL_JAAS_USERNAME);
@@ -109,7 +112,7 @@ public void initialize(AuthenticateContext context) throws AuthenticationExcepti
         }
 
         try {
-            saslClient = createSaslClient(mechanism, hostAddress, pros, loginManager);
+            saslClient = createSaslClient(mechanism, hostAddress, pros, loginManager, serviceName);
         } catch (Exception e) {
             throw new AuthenticationException("Failed to create SASL client", e);
         }

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

@@ -91,17 +91,18 @@ public static SaslServer createSaslServer(
     }
 
     public static SaslClient createSaslClient(
-            String mechanism, String hostAddress, Map<String, ?> props, LoginManager loginManager)
+            String mechanism,
+            String hostAddress,
+            Map<String, ?> props,
+            LoginManager loginManager,
+            String serviceName)
             throws PrivilegedActionException {
 
         return Subject.doAs(
                 loginManager.subject(),
                 (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={}",
                                     serviceName,