Fix for Cloudera CDS 2.2.0+ Spark credentialsRequired method (#1332)

Cloudera's CDS 2.2.0+ Spark fork introduced a non-passive method
signature that results in the inability for the ES hadoop integration
to work.  Cloudera's CDS fork is the main means for CDH5 consumers to
consume newer versions of Spark. CDS is only availble to be installed
via a Cloudera Manager's parcel and is distinctly different from the (elder)
version of Apache Spark that ships with CDH5. Supporting this fork is
challenging since it difficult difficult to test manually, and even more
so with the integration tests. CDS is for CDH5 consumers (likely the majority
at this time) that want to use Java 8 (likey the majority at this time),
and run newer versions of Spark (a very popular integration).

Related commit: https://github.com/cloudera/spark/commit/0972663ef3a87f4dcf2f4a936216bb5da0232b4e#diff-73c35ef723336affd463d9657df7583f

Fixes #1301

spark/sql-20/src/main/scala/org/elasticsearch/spark/deploy/yarn/security/EsServiceCredentialProvider.scala

@@ -75,12 +75,31 @@ class EsServiceCredentialProvider extends ServiceCredentialProvider {
   override def serviceName: String = "elasticsearch"
 
   /**
-   * Given a configuration, check to see if tokens would be required.
-   * @param hadoopConf the current configuration
-   * @return true if tokens should be gathered, false if they should not be
-   */
+    *  Given a configuration, check to see if tokens would be required.
+    *
+    * @param hadoopConf the current Hadoop configuration
+    * @return true if tokens should be gathered, false if they should not be
+    */
   override def credentialsRequired(hadoopConf: Configuration): Boolean = {
-    val settings = HadoopSettingsManager.loadFrom(hadoopConf)
+    credentialsRequired(null, hadoopConf)
+  }
+
+  /**
+    *  Given a configuration, check to see if tokens would be required.
+    *
+    * @param sparkConf the current Spark configuration - used by Cloudera's CDS Spark fork (#1301)
+    * @param hadoopConf the current Hadoop configuration
+    * @return true if tokens should be gathered, false if they should not be
+    */
+  def credentialsRequired(sparkConf: SparkConf, hadoopConf: Configuration): Boolean = {
+    val settings = if (sparkConf != null) {
+      new CompositeSettings(util.Arrays.asList(
+        new SparkSettingsManager().load(sparkConf),
+        new HadoopSettingsManager().load(hadoopConf)
+      ))
+    } else {
+      HadoopSettingsManager.loadFrom(hadoopConf)
+    }
     val isSecurityEnabled = UserGroupInformation.isSecurityEnabled
     val esAuthMethod = settings.getSecurityAuthenticationMethod
     val required = isSecurityEnabled && AuthenticationMethod.KERBEROS.equals(esAuthMethod)