[SPARK-39949][SS] Use canonical host name for principals in KafkaTest…

…Utils

### What changes were proposed in this pull request?

This PR proposes to use canonical host name for principals in KafkaTestUtils.

This is effectively reverting a part of SPARK-39530, as we used the canonical host name before SPARK-39530. (We used IP as "127.0.0.1" (loopback) and the host part of principal as the canonical host name of "127.0.0.1") Picking up the actual IP to use IPv6 if appropriate remains unchanged from SPARK-39530.

### Why are the changes needed?

KafkaDelegationTokenSuite fails on AWS EC2.

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

No.

### How was this patch tested?

Ran KafkaDelegationTokenSuite in AWS EC2 via following command:

```
build/sbt "sql-kafka-0-10/testOnly *KafkaDelegationTokenSuite"
```

This suite failed on current master branch and succeeds with this PR.

kerberos debug message on failure (only picked up the last error part as it's quite long and I need to redact):

```
>>>KRBError:
	 sTime is Tue Aug 02 02:30:29 PDT 2022 1659432629000
KrbException: Server not found in Kerberos database (7) - Server not found in Kerberos database
  | => s suSec is 100
	 error code is 7
	 error Message is Server not found in Kerberos database
	 crealm is EXAMPLE.COM
	 sname is zookeeper/ip-<ipv4-dot-replaced-with-bar>.<region>.compute.internalEXAMPLE.COM
	 msgType is 30
	at sun.security.krb5.KrbTgsRep.<init>(KrbTgsRep.java:73)
  | => sat sun.security.krb5.KrbTgsReq.getReply(KrbTgsReq.java:226)
	at sun.security.krb5.KrbTgsReq.sendAndGetCreds(KrbTgsReq.java:237)
	at sun.security.krb5.internal.CredentialsUtil.serviceCredsSingle(CredentialsUtil.java:477)
	at sun.security.krb5.internal.CredentialsUtil.serviceCreds(CredentialsUtil.java:340)
	at sun.security.krb5.internal.CredentialsUtil.serviceCreds(CredentialsUtil.java:314)
	at sun.security.krb5.internal.CredentialsUtil.acquireServiceCreds(CredentialsUtil.java:169)
	at sun.security.krb5.Credentials.acquireServiceCreds(Credentials.java:490)
	at sun.security.jgss.krb5.Krb5Context.initSecContext(Krb5Context.java:695)
	at sun.security.jgss.GSSContextImpl.initSecContext(GSSContextImpl.java:248)
	at sun.security.jgss.GSSContextImpl.initSecContext(GSSContextImpl.java:179)
	at com.sun.security.sasl.gsskerb.GssKrb5Client.evaluateChallenge(GssKrb5Client.java:192)
	at org.apache.zookeeper.client.ZooKeeperSaslClient$1.run(ZooKeeperSaslClient.java:320)
	at org.apache.zookeeper.client.ZooKeeperSaslClient$1.run(ZooKeeperSaslClient.java:317)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.zookeeper.client.ZooKeeperSaslClient.createSaslToken(ZooKeeperSaslClient.java:317)
	at org.apache.zookeeper.client.ZooKeeperSaslClient.createSaslToken(ZooKeeperSaslClient.java:303)
	at org.apache.zookeeper.client.ZooKeeperSaslClient.sendSaslPacket(ZooKeeperSaslClient.java:366)
	at org.apache.zookeeper.client.ZooKeeperSaslClient.initialize(ZooKeeperSaslClient.java:403)
	at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1220)
Caused by: KrbException: Identifier doesn't match expected value (906)
	at sun.security.krb5.internal.KDCRep.init(KDCRep.java:140)
	at sun.security.krb5.internal.TGSRep.init(TGSRep.java:65)
	at sun.security.krb5.internal.TGSRep.<init>(TGSRep.java:60)
	at sun.security.krb5.KrbTgsRep.<init>(KrbTgsRep.java:55)
	... 20 more
```

Closes apache#37377 from HeartSaVioR/SPARK-39949.

Authored-by: Jungtaek Lim <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala

@@ -71,6 +71,11 @@ class KafkaTestUtils(
   private val localHostNameForURI = Utils.localHostNameForURI()
   logInfo(s"Local host name is $localHostNameForURI")
 
+  // MiniKDC uses canonical host name on host part, hence we need to provide canonical host name
+  // on the 'host' part of the principal.
+  private val localCanonicalHostName = Utils.localCanonicalHostName()
+  logInfo(s"Local canonical host name is $localCanonicalHostName")
+
   private var kdc: MiniKdc = _
 
   // Zookeeper related configurations
@@ -88,7 +93,7 @@ class KafkaTestUtils(
   private var brokerConf: KafkaConfig = _
 
   private val brokerServiceName = "kafka"
-  private val clientUser = s"client/$localHostNameForURI"
+  private val clientUser = s"client/$localCanonicalHostName"
   private var clientKeytabFile: File = _
 
   // Kafka broker server
@@ -202,17 +207,17 @@ class KafkaTestUtils(
     assert(kdcReady, "KDC should be set up beforehand")
     val baseDir = Utils.createTempDir()
 
-    val zkServerUser = s"zookeeper/$localHostNameForURI"
+    val zkServerUser = s"zookeeper/$localCanonicalHostName"
     val zkServerKeytabFile = new File(baseDir, "zookeeper.keytab")
     kdc.createPrincipal(zkServerKeytabFile, zkServerUser)
     logDebug(s"Created keytab file: ${zkServerKeytabFile.getAbsolutePath()}")
 
-    val zkClientUser = s"zkclient/$localHostNameForURI"
+    val zkClientUser = s"zkclient/$localCanonicalHostName"
     val zkClientKeytabFile = new File(baseDir, "zkclient.keytab")
     kdc.createPrincipal(zkClientKeytabFile, zkClientUser)
     logDebug(s"Created keytab file: ${zkClientKeytabFile.getAbsolutePath()}")
 
-    val kafkaServerUser = s"kafka/$localHostNameForURI"
+    val kafkaServerUser = s"kafka/$localCanonicalHostName"
     val kafkaServerKeytabFile = new File(baseDir, "kafka.keytab")
     kdc.createPrincipal(kafkaServerKeytabFile, kafkaServerUser)
     logDebug(s"Created keytab file: ${kafkaServerKeytabFile.getAbsolutePath()}")