This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 5e0f0da [SPARK-37391][SQL] JdbcConnectionProvider tells if it
modifies security context
5e0f0da is described below
commit 5e0f0da58edc5ce60fa972515fba73655400543d
Author: Danny Guinther <[email protected]>
AuthorDate: Fri Dec 24 10:48:04 2021 +0900
[SPARK-37391][SQL] JdbcConnectionProvider tells if it modifies security
context
# branch-3.1 version!
For master version see : https://github.com/apache/spark/pull/34745
Augments the JdbcConnectionProvider API such that a provider can indicate
that it will need to modify the global security configuration when establishing
a connection, and as such, if access to the global security configuration
should be synchronized to prevent races.
### What changes were proposed in this pull request?
As suggested by gaborgsomogyi
[here](https://github.com/apache/spark/pull/29024/files#r755788709), augments
the `JdbcConnectionProvider` API to include a `modifiesSecurityContext` method
that can be used by `ConnectionProvider` to determine when
`SecurityConfigurationLock.synchronized` is required to avoid race conditions
when establishing a JDBC connection.
### Why are the changes needed?
Provides a path forward for working around a significant bottleneck
introduced by synchronizing `SecurityConfigurationLock` every time a connection
is established. The synchronization isn't always needed and it should be at the
discretion of the `JdbcConnectionProvider` to determine when locking is
necessary. See [SPARK-37391](https://issues.apache.org/jira/browse/SPARK-37391)
or [this thread](https://github.com/apache/spark/pull/29024/files#r754441783).
### Does this PR introduce _any_ user-facing change?
Any existing implementations of `JdbcConnectionProvider` will need to add a
definition of `modifiesSecurityContext`. I'm also open to adding a default
implementation, but it seemed to me that requiring an explicit implementation
of the method was preferable.
A drop-in implementation that would continue the existing behavior is:
```scala
override def modifiesSecurityContext(
driver: Driver,
options: Map[String, String]
): Boolean = true
```
### How was this patch tested?
Unit tests. Also ran a real workflow by swapping in a locally published
version of `spark-sql` into my local spark 3.1.2 installation's jars.
Closes #34988 from
tdg5/SPARK-37391-opt-in-security-configuration-sync-branch-3.1.
Authored-by: Danny Guinther <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
project/MimaExcludes.scala | 5 ++++-
.../jdbc/connection/BasicConnectionProvider.scala | 8 ++++++++
.../jdbc/connection/ConnectionProvider.scala | 23 +++++++++++++---------
.../spark/sql/jdbc/JdbcConnectionProvider.scala | 19 ++++++++++++++++--
.../IntentionallyFaultyConnectionProvider.scala | 4 ++++
5 files changed, 47 insertions(+), 12 deletions(-)
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index c29dd9e..c95c3815 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -105,7 +105,10 @@ object MimaExcludes {
ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("org.apache.spark.ml.classification.BinaryLogisticRegressionSummary.weightCol"),
// [SPARK-32879] Pass SparkSession.Builder options explicitly to
SparkSession
-
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.SparkSession.this")
+
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.SparkSession.this"),
+
+ // [SPARK-37391][SQL] JdbcConnectionProvider tells if it modifies security
context
+
ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.jdbc.JdbcConnectionProvider.modifiesSecurityContext")
)
// Exclude rules for 3.0.x
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/BasicConnectionProvider.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/BasicConnectionProvider.scala
index 890205f..84b1ab9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/BasicConnectionProvider.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/BasicConnectionProvider.scala
@@ -48,4 +48,12 @@ private[jdbc] class BasicConnectionProvider extends
JdbcConnectionProvider with
logDebug(s"JDBC connection initiated with URL: ${jdbcOptions.url} and
properties: $properties")
driver.connect(jdbcOptions.url, properties)
}
+
+ override def modifiesSecurityContext(
+ driver: Driver,
+ options: Map[String, String]
+ ): Boolean = {
+ // BasicConnectionProvider is the default unsecure connection provider, so
just return false
+ false
+ }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/ConnectionProvider.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/ConnectionProvider.scala
index fbc6970..e54d8a9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/ConnectionProvider.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/ConnectionProvider.scala
@@ -60,16 +60,21 @@ private[jdbc] object ConnectionProvider extends Logging {
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 {
- // Inside getConnection it's safe to get parent again because
SecurityConfigurationLock
- // makes sure it's untouched
- val parent = Configuration.getConfiguration
- try {
- filteredProviders.head.getConnection(driver, options)
- } finally {
- logDebug("Restoring original security configuration")
- Configuration.setConfiguration(parent)
+ val selectedProvider = filteredProviders.head
+ if (selectedProvider.modifiesSecurityContext(driver, options)) {
+ SecurityConfigurationLock.synchronized {
+ // Inside getConnection it's safe to get parent again because
SecurityConfigurationLock
+ // makes sure it's untouched
+ val parent = Configuration.getConfiguration
+ try {
+ selectedProvider.getConnection(driver, options)
+ } finally {
+ logDebug("Restoring original security configuration")
+ Configuration.setConfiguration(parent)
+ }
}
+ } else {
+ selectedProvider.getConnection(driver, options)
}
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcConnectionProvider.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcConnectionProvider.scala
index 1e8abca..10eebf3 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcConnectionProvider.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcConnectionProvider.scala
@@ -53,12 +53,27 @@ abstract class JdbcConnectionProvider {
def canHandle(driver: Driver, options: Map[String, String]): Boolean
/**
- * Opens connection toward the database. Since global JVM security
configuration change may needed
- * this API is called synchronized by `SecurityConfigurationLock` to avoid
race.
+ * Opens connection to the database. Since global JVM security configuration
change may be
+ * needed this API is called synchronized by `SecurityConfigurationLock` to
avoid race when
+ * `modifiesSecurityContext` returns true for the given driver with the
given options.
*
* @param driver Java driver which initiates the connection
* @param options Driver options which initiates the connection
* @return a `Connection` object that represents a connection to the URL
*/
def getConnection(driver: Driver, options: Map[String, String]): Connection
+
+ /**
+ * Checks if this connection provider instance needs to modify global
security configuration to
+ * handle authentication and thus should synchronize access to the security
configuration while
+ * the given driver is initiating a connection with the given options.
+ *
+ * @param driver Java driver which initiates the connection
+ * @param options Driver options which initiates the connection
+ * @return True if the connection provider will need to modify the security
configuration when
+ * initiating a connection with the given driver with the given options.
+ *
+ * @since 3.1.3
+ */
+ def modifiesSecurityContext(driver: Driver, options: Map[String, String]):
Boolean
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/IntentionallyFaultyConnectionProvider.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/IntentionallyFaultyConnectionProvider.scala
index 329d79c..f5d6d34 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/IntentionallyFaultyConnectionProvider.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/jdbc/connection/IntentionallyFaultyConnectionProvider.scala
@@ -27,6 +27,10 @@ private class IntentionallyFaultyConnectionProvider extends
JdbcConnectionProvid
override val name: String = "IntentionallyFaultyConnectionProvider"
override def canHandle(driver: Driver, options: Map[String, String]):
Boolean = true
override def getConnection(driver: Driver, options: Map[String, String]):
Connection = null
+ override def modifiesSecurityContext(
+ driver: Driver,
+ options: Map[String, String]
+ ): Boolean = false
}
private object IntentionallyFaultyConnectionProvider {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
