[SPARK-32001][SQL]Create JDBC authentication provider developer API

core/src/main/scala/org/apache/spark/security/SecurityConfigurationLock.scala

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security
+
+/**
+ * There are cases when global JVM security configuration must be modified.
+ * In order to avoid race the modification must be synchronized with this.
+ */
+object SecurityConfigurationLock

external/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/DB2KrbIntegrationSuite.scala

@@ -54,7 +54,7 @@ class DB2KrbIntegrationSuite extends DockerKrbJDBCIntegrationSuite {
         JDBCOptions.JDBC_KEYTAB -> keytabFileName,
         JDBCOptions.JDBC_PRINCIPAL -> principal
       ))
-      new DB2ConnectionProvider(null, options).getAdditionalProperties()
+      new DB2ConnectionProvider().getAdditionalProperties(options)
     }
 
     override def beforeContainerStart(

sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.jdbc.JdbcConnectionProvider

@@ -0,0 +1,6 @@
+org.apache.spark.sql.execution.datasources.jdbc.connection.BasicConnectionProvider
+org.apache.spark.sql.execution.datasources.jdbc.connection.DB2ConnectionProvider
+org.apache.spark.sql.execution.datasources.jdbc.connection.MariaDBConnectionProvider
+org.apache.spark.sql.execution.datasources.jdbc.connection.MSSQLConnectionProvider
+org.apache.spark.sql.execution.datasources.jdbc.connection.PostgresConnectionProvider
+org.apache.spark.sql.execution.datasources.jdbc.connection.OracleConnectionProvider

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala

@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
  * Options for the JDBC data source.
  */
 class JDBCOptions(
-    @transient val parameters: CaseInsensitiveMap[String])
+    val parameters: CaseInsensitiveMap[String])
   extends Serializable with Logging {
 
   import JDBCOptions._
@@ -209,7 +209,7 @@ class JDBCOptions(
 }
 
 class JdbcOptionsInWrite(
-    @transient override val parameters: CaseInsensitiveMap[String])
+    override val parameters: CaseInsensitiveMap[String])
   extends JDBCOptions(parameters) {
 
   import JDBCOptions._

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -63,7 +63,7 @@ object JdbcUtils extends Logging {
         throw new IllegalStateException(
           s"Did not find registered driver with class $driverClass")
       }
-      val connection = ConnectionProvider.create(driver, options).getConnection()
+      val connection = ConnectionProvider.create(driver, options.parameters)
       require(connection != null,
         s"The driver could not open a JDBC connection. Check the URL: ${options.url}")
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/BasicConnectionProvider.scala

@@ -18,18 +18,30 @@
 package org.apache.spark.sql.execution.datasources.jdbc.connection
 
 import java.sql.{Connection, Driver}
+import java.util.Properties
 
-import scala.collection.JavaConverters._
-
+import org.apache.spark.internal.Logging
 import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
+import org.apache.spark.sql.jdbc.JdbcConnectionProvider
+
+private[jdbc] class BasicConnectionProvider extends JdbcConnectionProvider with Logging {
+  /**
+   * Additional properties for data connection (Data source property takes precedence).
+   */
+  def getAdditionalProperties(options: JDBCOptions): Properties = new Properties()
+
+  override def canHandle(driver: Driver, options: Map[String, String]): Boolean = {
+    val jdbcOptions = new JDBCOptions(options)
+    jdbcOptions.keytab == null || jdbcOptions.principal == null
+  }
 
-private[jdbc] class BasicConnectionProvider(driver: Driver, options: JDBCOptions)
-  extends ConnectionProvider {
-  def getConnection(): Connection = {
-    val properties = getAdditionalProperties()
-    options.asConnectionProperties.entrySet().asScala.foreach { e =>
-      properties.put(e.getKey(), e.getValue())
+  override def getConnection(driver: Driver, options: Map[String, String]): Connection = {
+    val jdbcOptions = new JDBCOptions(options)
+    val properties = getAdditionalProperties(jdbcOptions)
+    options.foreach { case(k, v) =>
+      properties.put(k, v)
     }
-    driver.connect(options.url, properties)
+    logDebug(s"JDBC connection initiated with URL: ${jdbcOptions.url} and properties: $properties")
+    driver.connect(jdbcOptions.url, properties)
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/ConnectionProvider.scala

@@ -18,60 +18,45 @@
 package org.apache.spark.sql.execution.datasources.jdbc.connection
 
 import java.sql.{Connection, Driver}
-import java.util.Properties
+import java.util.ServiceLoader
 
-import org.apache.spark.internal.Logging
-import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
-
-/**
- * Connection provider which opens connection toward various databases (database specific instance
- * needed). If kerberos authentication required then it's the provider's responsibility to set all
- * the parameters.
- */
-private[jdbc] trait ConnectionProvider {
-  /**
-   * Additional properties for data connection (Data source property takes precedence).
-   */
-  def getAdditionalProperties(): Properties = new Properties()
+import scala.collection.mutable
 
-  /**
-   * Opens connection toward the database.
-   */
-  def getConnection(): Connection
-}
+import org.apache.spark.internal.Logging
+import org.apache.spark.security.SecurityConfigurationLock
+import org.apache.spark.sql.jdbc.JdbcConnectionProvider
+import org.apache.spark.util.Utils
 
 private[jdbc] object ConnectionProvider extends Logging {
-  def create(driver: Driver, options: JDBCOptions): ConnectionProvider = {
-    if (options.keytab == null || options.principal == null) {
-      logDebug("No authentication configuration found, using basic connection provider")
-      new BasicConnectionProvider(driver, options)
-    } else {
-      logDebug("Authentication configuration found, using database specific connection provider")
-      options.driverClass match {
-        case PostgresConnectionProvider.driverClass =>
-          logDebug("Postgres connection provider found")
-          new PostgresConnectionProvider(driver, options)
-
-        case MariaDBConnectionProvider.driverClass =>
-          logDebug("MariaDB connection provider found")
-          new MariaDBConnectionProvider(driver, options)
-
-        case DB2ConnectionProvider.driverClass =>
-          logDebug("DB2 connection provider found")
-          new DB2ConnectionProvider(driver, options)
-
-        case MSSQLConnectionProvider.driverClass =>
-          logDebug("MS SQL connection provider found")
-          new MSSQLConnectionProvider(driver, options)
-
-        case OracleConnectionProvider.driverClass =>
-          logDebug("Oracle connection provider found")
-          new OracleConnectionProvider(driver, options)
-
-        case _ =>
-          throw new IllegalArgumentException(s"Driver ${options.driverClass} does not support " +
-            "Kerberos authentication")
+  private val providers = loadProviders()
+
+  def loadProviders(): Seq[JdbcConnectionProvider] = {
+    val loader = ServiceLoader.load(classOf[JdbcConnectionProvider],
+      Utils.getContextOrSparkClassLoader)
+    val providers = mutable.ArrayBuffer[JdbcConnectionProvider]()
+
+    val iterator = loader.iterator
+    while (iterator.hasNext) {
+      try {
+        val provider = iterator.next
+        logDebug(s"Loaded built in provider: $provider")
+        providers += provider
+      } catch {
+        case t: Throwable =>
+          logError(s"Failed to load built in provider.", t)
       }
     }
+    // Seems duplicate but it's needed for Scala 2.13
+    providers.toSeq
+  }
+
+  def create(driver: Driver, options: Map[String, String]): Connection = {
+    val filteredProviders = providers.filter(_.canHandle(driver, options))
+    require(filteredProviders.size == 1,
+      "JDBC connection initiated but not exactly one connection provider found which can handle " +
+        s"it. Found active providers: ${filteredProviders.mkString(", ")}")
+    SecurityConfigurationLock.synchronized {
+      filteredProviders.head.getConnection(driver, options)
+    }
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/DB2ConnectionProvider.scala

@@ -25,37 +25,36 @@ import org.apache.hadoop.security.UserGroupInformation
 
 import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
 
-private[sql] class DB2ConnectionProvider(driver: Driver, options: JDBCOptions)
-  extends SecureConnectionProvider(driver, options) {
-  override val appEntry: String = "JaasClient"
-
-  override def getConnection(): Connection = {
-    setAuthenticationConfigIfNeeded()
-    UserGroupInformation.loginUserFromKeytabAndReturnUGI(options.principal, options.keytab).doAs(
-      new PrivilegedExceptionAction[Connection]() {
-        override def run(): Connection = {
-          DB2ConnectionProvider.super.getConnection()
+private[sql] class DB2ConnectionProvider extends SecureConnectionProvider {
+  override val driverClass = "com.ibm.db2.jcc.DB2Driver"
+
+  override def appEntry(driver: Driver, options: JDBCOptions): String = "JaasClient"
+
+  override def getConnection(driver: Driver, options: Map[String, String]): Connection = {
+    val jdbcOptions = new JDBCOptions(options)
+    setAuthenticationConfigIfNeeded(driver, jdbcOptions)
+    UserGroupInformation.loginUserFromKeytabAndReturnUGI(jdbcOptions.principal, jdbcOptions.keytab)
+      .doAs(
+        new PrivilegedExceptionAction[Connection]() {
+          override def run(): Connection = {
+            DB2ConnectionProvider.super.getConnection(driver, options)
+          }
         }
-      }
-    )
+      )
   }
 
-  override def getAdditionalProperties(): Properties = {
+  override def getAdditionalProperties(options: JDBCOptions): Properties = {
     val result = new Properties()
     // 11 is the integer value for kerberos
     result.put("securityMechanism", new String("11"))
     result.put("KerberosServerPrincipal", options.principal)
     result
   }
 
-  override def setAuthenticationConfigIfNeeded(): Unit = SecurityConfigurationLock.synchronized {
-    val (parent, configEntry) = getConfigWithAppEntry()
+  override def setAuthenticationConfigIfNeeded(driver: Driver, options: JDBCOptions): Unit = {
+    val (parent, configEntry) = getConfigWithAppEntry(driver, options)
     if (configEntry == null || configEntry.isEmpty) {
-      setAuthenticationConfig(parent)
+      setAuthenticationConfig(parent, driver, options)
     }
   }
 }
-
-private[sql] object DB2ConnectionProvider {
-  val driverClass = "com.ibm.db2.jcc.DB2Driver"
-}

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/MSSQLConnectionProvider.scala

@@ -25,12 +25,11 @@ import org.apache.hadoop.security.UserGroupInformation
 
 import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
 
-private[sql] class MSSQLConnectionProvider(
-    driver: Driver,
-    options: JDBCOptions,
-    parserMethod: String = "parseAndMergeProperties"
-  ) extends SecureConnectionProvider(driver, options) {
-  override val appEntry: String = {
+private[sql] class MSSQLConnectionProvider extends SecureConnectionProvider {
+  override val driverClass = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+  val parserMethod: String = "parseAndMergeProperties"
+
+  override def appEntry(driver: Driver, options: JDBCOptions): String = {
     val configName = "jaasConfigurationName"
     val appEntryDefault = "SQLJDBCDriver"
 
@@ -58,27 +57,29 @@ private[sql] class MSSQLConnectionProvider(
     }
   }
 
-  override def getConnection(): Connection = {
-    setAuthenticationConfigIfNeeded()
-    UserGroupInformation.loginUserFromKeytabAndReturnUGI(options.principal, options.keytab).doAs(
-      new PrivilegedExceptionAction[Connection]() {
-        override def run(): Connection = {
-          MSSQLConnectionProvider.super.getConnection()
+  override def getConnection(driver: Driver, options: Map[String, String]): Connection = {
+    val jdbcOptions = new JDBCOptions(options)
+    setAuthenticationConfigIfNeeded(driver, jdbcOptions)
+    UserGroupInformation.loginUserFromKeytabAndReturnUGI(jdbcOptions.principal, jdbcOptions.keytab)
+      .doAs(
+        new PrivilegedExceptionAction[Connection]() {
+          override def run(): Connection = {
+            MSSQLConnectionProvider.super.getConnection(driver, options)
+          }
         }
-      }
-    )
+      )
   }
 
-  override def getAdditionalProperties(): Properties = {
+  override def getAdditionalProperties(options: JDBCOptions): Properties = {
     val result = new Properties()
     // These props needed to reach internal kerberos authentication in the JDBC driver
     result.put("integratedSecurity", "true")
     result.put("authenticationScheme", "JavaKerberos")
     result
   }
 
-  override def setAuthenticationConfigIfNeeded(): Unit = SecurityConfigurationLock.synchronized {
-    val (parent, configEntry) = getConfigWithAppEntry()
+  override def setAuthenticationConfigIfNeeded(driver: Driver, options: JDBCOptions): Unit = {
+    val (parent, configEntry) = getConfigWithAppEntry(driver, options)
     /**
      * Couple of things to mention here (v8.2.2 client):
      * 1. MS SQL supports JAAS application name configuration
@@ -87,11 +88,7 @@ private[sql] class MSSQLConnectionProvider(
     val entryUsesKeytab = configEntry != null &&
       configEntry.exists(_.getOptions().get("useKeyTab") == "true")
     if (configEntry == null || configEntry.isEmpty || !entryUsesKeytab) {
-      setAuthenticationConfig(parent)
+      setAuthenticationConfig(parent, driver, options)
     }
   }
 }
-
-private[sql] object MSSQLConnectionProvider {
-  val driverClass = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
-}

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/MariaDBConnectionProvider.scala

@@ -21,14 +21,14 @@ import java.sql.Driver
 
 import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
 
-private[jdbc] class MariaDBConnectionProvider(driver: Driver, options: JDBCOptions)
-  extends SecureConnectionProvider(driver, options) {
-  override val appEntry: String = {
+private[jdbc] class MariaDBConnectionProvider extends SecureConnectionProvider {
+  override val driverClass = "org.mariadb.jdbc.Driver"
+
+  override def appEntry(driver: Driver, options: JDBCOptions): String =
     "Krb5ConnectorContext"
-  }
 
-  override def setAuthenticationConfigIfNeeded(): Unit = SecurityConfigurationLock.synchronized {
-    val (parent, configEntry) = getConfigWithAppEntry()
+  override def setAuthenticationConfigIfNeeded(driver: Driver, options: JDBCOptions): Unit = {
+    val (parent, configEntry) = getConfigWithAppEntry(driver, options)
     /**
      * Couple of things to mention here (v2.5.4 client):
      * 1. MariaDB doesn't support JAAS application name configuration
@@ -37,11 +37,7 @@ private[jdbc] class MariaDBConnectionProvider(driver: Driver, options: JDBCOptio
     val entryUsesKeytab = configEntry != null &&
       configEntry.exists(_.getOptions().get("useKeyTab") == "true")
     if (configEntry == null || configEntry.isEmpty || !entryUsesKeytab) {
-      setAuthenticationConfig(parent)
+      setAuthenticationConfig(parent, driver, options)
     }
   }
 }
-
-private[sql] object MariaDBConnectionProvider {
-  val driverClass = "org.mariadb.jdbc.Driver"
-}