Review fixes:

 * Reordered TokenUtil methods to make it more readable.
 * Enhanced additional parameters doc to make things more explicit.

Author: gaborgsomogyi

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -651,37 +651,38 @@ package object config {
   private[spark] val KAFKA_BOOTSTRAP_SERVERS =
     ConfigBuilder("spark.kafka.bootstrap.servers")
       .doc("A list of coma separated host/port pairs to use for establishing the initial " +
-        "connection to the Kafka cluster. For further details please see kafka documentation.")
+        "connection to the Kafka cluster. For further details please see kafka documentation. " +
+        "Only used to obtain delegation token.")
       .stringConf
       .createOptional
 
   private[spark] val KAFKA_SECURITY_PROTOCOL =
     ConfigBuilder("spark.kafka.security.protocol")
       .doc("Protocol used to communicate with brokers. For further details please see kafka " +
-        "documentation.")
+        "documentation. Only used to obtain delegation token.")
       .stringConf
       .createWithDefault("SASL_SSL")
 
   private[spark] val KAFKA_KERBEROS_SERVICE_NAME =
     ConfigBuilder("spark.kafka.sasl.kerberos.service.name")
       .doc("The Kerberos principal name that Kafka runs as. This can be defined either in " +
         "Kafka's JAAS config or in Kafka's config. For further details please see kafka " +
-        "documentation.")
+        "documentation. Only used to obtain delegation token.")
       .stringConf
       .createOptional
 
   private[spark] val KAFKA_TRUSTSTORE_LOCATION =
     ConfigBuilder("spark.kafka.ssl.truststore.location")
       .doc("The location of the trust store file. For further details please see kafka " +
-        "documentation.")
+        "documentation. Only used to obtain delegation token.")
       .stringConf
       .createOptional
 
   private[spark] val KAFKA_TRUSTSTORE_PASSWORD =
     ConfigBuilder("spark.kafka.ssl.truststore.password")
       .doc("The store password for the key store file. This is optional for client and only " +
         "needed if ssl.keystore.location is configured. For further details please see kafka " +
-        "documentation.")
+        "documentation. Only used to obtain delegation token.")
       .stringConf
       .createOptional
 }

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/TokenUtil.scala

@@ -40,20 +40,19 @@ private[kafka010] object TokenUtil extends Logging {
     override def getKind: Text = TOKEN_KIND;
   }
 
-  private def printToken(token: DelegationToken): Unit = {
-    if (log.isDebugEnabled) {
-      val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm")
-      logDebug("%-15s %-30s %-15s %-25s %-15s %-15s %-15s".format(
-        "TOKENID", "HMAC", "OWNER", "RENEWERS", "ISSUEDATE", "EXPIRYDATE", "MAXDATE"))
-      val tokenInfo = token.tokenInfo
-      logDebug("%-15s [hidden] %-15s %-25s %-15s %-15s %-15s".format(
-        tokenInfo.tokenId,
-        tokenInfo.owner,
-        tokenInfo.renewersAsString,
-        dateFormat.format(tokenInfo.issueTimestamp),
-        dateFormat.format(tokenInfo.expiryTimestamp),
-        dateFormat.format(tokenInfo.maxTimestamp)))
-    }
+  def obtainToken(sparkConf: SparkConf): (Token[_ <: TokenIdentifier], Long) = {
+    val adminClient = AdminClient.create(createAdminClientProperties(sparkConf))
+    val createDelegationTokenOptions = new CreateDelegationTokenOptions()
+    val createResult = adminClient.createDelegationToken(createDelegationTokenOptions)
+    val token = createResult.delegationToken().get()
+    printToken(token)
+
+    new Token[KafkaDelegationTokenIdentifier](
+      token.tokenInfo.tokenId.getBytes,
+      token.hmacAsBase64String.getBytes,
+      TOKEN_KIND,
+      TOKEN_SERVICE
+    )
   }
 
   private[kafka010] def createAdminClientProperties(sparkConf: SparkConf): Properties = {
@@ -94,18 +93,19 @@ private[kafka010] object TokenUtil extends Logging {
     adminClientProperties
   }
 
-  def obtainToken(sparkConf: SparkConf): Token[_ <: TokenIdentifier] = {
-    val adminClient = AdminClient.create(createAdminClientProperties(sparkConf))
-    val createDelegationTokenOptions = new CreateDelegationTokenOptions()
-    val createResult = adminClient.createDelegationToken(createDelegationTokenOptions)
-    val token = createResult.delegationToken().get()
-    printToken(token)
-
-    new Token[KafkaDelegationTokenIdentifier](
-      token.tokenInfo.tokenId.getBytes,
-      token.hmacAsBase64String.getBytes,
-      TOKEN_KIND,
-      TOKEN_SERVICE
-    )
+  private def printToken(token: DelegationToken): Unit = {
+    if (log.isDebugEnabled) {
+      val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm")
+      logDebug("%-15s %-30s %-15s %-25s %-15s %-15s %-15s".format(
+        "TOKENID", "HMAC", "OWNER", "RENEWERS", "ISSUEDATE", "EXPIRYDATE", "MAXDATE"))
+      val tokenInfo = token.tokenInfo
+      logDebug("%-15s [hidden] %-15s %-25s %-15s %-15s %-15s".format(
+        tokenInfo.tokenId,
+        tokenInfo.owner,
+        tokenInfo.renewersAsString,
+        dateFormat.format(tokenInfo.issueTimestamp),
+        dateFormat.format(tokenInfo.expiryTimestamp),
+        dateFormat.format(tokenInfo.maxTimestamp)))
+    }
   }
 }