[KYUUBI #5961] Support to specify client kerberosAuthType as fromTick…

…etCache

# 🔍 Description

The use case:
- 1. the user code running in flink cluster and wrapped by transparent `ugi.doAs`
- 2. the current UserGroupInfomation is the proxy user not login/real user
- 3. user specify url with kyuubiClientTicketCache but does not work
- 4. the proxy UserGroupInfomation is used and then throw GSS issue.
```
Caused by: org.apache.kyuubi.shade.org.apache.thrift.transport.TTransportException: GSS initiate failed
    at org.apache.kyuubi.shade.org.apache.thrift.transport.TSaslTransport.sendAndThrowMessage(TSaslTransport.java:232)
    at org.apache.kyuubi.shade.org.apache.thrift.transport.TSaslTransport.open(TSaslTransport.java:316)
    at org.apache.kyuubi.shade.org.apache.thrift.transport.TSaslClientTransport.open(TSaslClientTransport.java:37)
    at org.apache.kyuubi.jdbc.hive.auth.TSubjectTransport.lambda$open$0(TSubjectTransport.java:47)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:422)
    at org.apache.kyuubi.jdbc.hive.auth.TSubjectTransport.open(TSubjectTransport.java:42)
    at org.apache.kyuubi.jdbc.hive.KyuubiConnection.openTransport(KyuubiConnection.java:458)
    at org.apache.kyuubi.jdbc.hive.KyuubiConnection.<init>(KyuubiConnection.java:207)
    ... 22 more
```

The root cause is that, for this case, the result of `isHadoopUserGroupInformationDoAs` is true.

So, `isFromSubjectAuthMode` is true.

In this pr, I want to specify the kerberosAuthType to fromTicketCache, and do not check `isHadoopUserGroupInformationDoAs`.

After this pr, customer can specify `kerberosAuthType=fromTicketCache` to leverage ticket cache prefer than `ugi.doAs`.

## Issue References 🔗

This pull request fixes #

## Describe Your Solution 🔧

Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context. List any dependencies that are required for this change.

## Types of changes 🔖

- [ ] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Test Plan 🧪

#### Behavior Without This Pull Request ⚰️

#### Behavior With This Pull Request 🎉

#### Related Unit Tests

---

# Checklist 📝

- [ ] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html)

**Be nice. Be informative.**

Closes #5961 from turboFei/warn_ticket_cache.

Closes #5961

6e06500 [Fei Wang] exception
c01a99e [Fei Wang] refine
098a37b [Fei Wang] do not infer fromSubject if fromTicketCache && ugi
042fa22 [Fei Wang] warn

Authored-by: Fei Wang <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
(cherry picked from commit 54086b0)
Signed-off-by: Cheng Pan <[email protected]>

kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/JdbcConnectionParams.java

@@ -52,7 +52,9 @@ public class JdbcConnectionParams {
   public static final String AUTH_KYUUBI_CLIENT_TICKET_CACHE = "kyuubiClientTicketCache";
   public static final String AUTH_PASSWD = "password";
   public static final String AUTH_KERBEROS_AUTH_TYPE = "kerberosAuthType";
+  public static final String AUTH_KERBEROS_AUTH_TYPE_FROM_KEYTAB = "fromKeytab";
   public static final String AUTH_KERBEROS_AUTH_TYPE_FROM_SUBJECT = "fromSubject";
+  public static final String AUTH_KERBEROS_AUTH_TYPE_FROM_TICKET_CACHE = "fromTicketCache";
   public static final String ANONYMOUS_USER = "anonymous";
   public static final String ANONYMOUS_PASSWD = "anonymous";
   public static final String USE_SSL = "ssl";

kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiConnection.java

@@ -843,28 +843,68 @@ private boolean isHadoopUserGroupInformationDoAs() {
     }
   }
 
+  private boolean isForciblyFromKeytabAuthMode() {
+    return AUTH_KERBEROS_AUTH_TYPE_FROM_KEYTAB.equalsIgnoreCase(
+        sessConfMap.get(AUTH_KERBEROS_AUTH_TYPE));
+  }
+
+  private boolean isForciblyFromSubjectAuthMode() {
+    return AUTH_KERBEROS_AUTH_TYPE_FROM_SUBJECT.equalsIgnoreCase(
+        sessConfMap.get(AUTH_KERBEROS_AUTH_TYPE));
+  }
+
+  private boolean isForciblyTgtCacheAuthMode() {
+    return AUTH_KERBEROS_AUTH_TYPE_FROM_TICKET_CACHE.equalsIgnoreCase(
+        sessConfMap.get(AUTH_KERBEROS_AUTH_TYPE));
+  }
+
   private boolean isKeytabAuthMode() {
-    return isSaslAuthMode()
-        && hasSessionValue(AUTH_PRINCIPAL)
+    // handle explicit cases first
+    if (isForciblyFromSubjectAuthMode() || isForciblyTgtCacheAuthMode()) {
+      return false;
+    }
+    if (isKerberosAuthMode() && isForciblyFromKeytabAuthMode()) {
+      return true;
+    }
+    if (isKerberosAuthMode()
+        && hasSessionValue(AUTH_KYUUBI_CLIENT_KEYTAB)
+        && !hasSessionValue(AUTH_KYUUBI_CLIENT_PRINCIPAL)) {
+      throw new IllegalArgumentException(
+          AUTH_KYUUBI_CLIENT_KEYTAB
+              + " is set but "
+              + AUTH_KYUUBI_CLIENT_PRINCIPAL
+              + " is not set");
+    }
+    // handle implicit cases then
+    return isKerberosAuthMode()
         && hasSessionValue(AUTH_KYUUBI_CLIENT_PRINCIPAL)
         && hasSessionValue(AUTH_KYUUBI_CLIENT_KEYTAB);
   }
 
   private boolean isFromSubjectAuthMode() {
-    return isSaslAuthMode()
-        && hasSessionValue(AUTH_PRINCIPAL)
-        && !hasSessionValue(AUTH_KYUUBI_CLIENT_PRINCIPAL)
+    // handle explicit cases first
+    if (isForciblyFromKeytabAuthMode() || isForciblyTgtCacheAuthMode()) {
+      return false;
+    }
+    if (isKerberosAuthMode() && isForciblyFromSubjectAuthMode()) {
+      return true;
+    }
+    // handle implicit cases then
+    return isKerberosAuthMode()
         && !hasSessionValue(AUTH_KYUUBI_CLIENT_KEYTAB)
-        && (AUTH_KERBEROS_AUTH_TYPE_FROM_SUBJECT.equalsIgnoreCase(
-                sessConfMap.get(AUTH_KERBEROS_AUTH_TYPE))
-            || isHadoopUserGroupInformationDoAs());
+        && isHadoopUserGroupInformationDoAs();
   }
 
   private boolean isTgtCacheAuthMode() {
-    return isSaslAuthMode()
-        && hasSessionValue(AUTH_PRINCIPAL)
-        && !hasSessionValue(AUTH_KYUUBI_CLIENT_PRINCIPAL)
-        && !hasSessionValue(AUTH_KYUUBI_CLIENT_KEYTAB);
+    // handle explicit cases first
+    if (isForciblyFromKeytabAuthMode() || isForciblyFromSubjectAuthMode()) {
+      return false;
+    }
+    if (isKerberosAuthMode() && isForciblyTgtCacheAuthMode()) {
+      return true;
+    }
+    // handle implicit cases then
+    return isKerberosAuthMode() && !hasSessionValue(AUTH_KYUUBI_CLIENT_KEYTAB);
   }
 
   private boolean isPlainSaslAuthMode() {

kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/auth/KerberosAuthentication.java

@@ -28,6 +28,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.InetAddress;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Map;
@@ -106,6 +107,9 @@ private static Configuration createLoginFromTgtCacheConfiguration(String ticketC
     if (StringUtils.isBlank(ticketCache)) {
       ticketCache = System.getenv("KRB5CCNAME");
     }
+    if (!Files.exists(Paths.get(ticketCache))) {
+      LOG.warn("TicketCache {} does not exist", ticketCache);
+    }
     if (StringUtils.isNotBlank(ticketCache)) {
       optionsBuilder.put("ticketCache", ticketCache);
     }