This is an automated email from the ASF dual-hosted git repository. vanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new 773efed [SPARK-26254][CORE] Extract Hive + Kafka dependencies from Core. 773efed is described below commit 773efede2080639ef7eb546c11bf81401d160db2 Author: Gabor Somogyi <gabor.g.somo...@gmail.com> AuthorDate: Fri Jan 25 10:36:00 2019 -0800 [SPARK-26254][CORE] Extract Hive + Kafka dependencies from Core. ## What changes were proposed in this pull request? There are ugly provided dependencies inside core for the following: * Hive * Kafka In this PR I've extracted them out. This PR contains the following: * Token providers are now loaded with service loader * Hive token provider moved to hive project * Kafka token provider extracted into a new project ## How was this patch tested? Existing + newly added unit tests. Additionally tested on cluster. Closes #23499 from gaborgsomogyi/SPARK-26254. Authored-by: Gabor Somogyi <gabor.g.somo...@gmail.com> Signed-off-by: Marcelo Vanzin <van...@cloudera.com> --- core/pom.xml | 40 -------- ...k.deploy.security.HadoopDelegationTokenProvider | 2 + .../security/HBaseDelegationTokenProvider.scala | 2 + .../security/HadoopDelegationTokenManager.scala | 38 ++++---- .../security/HadoopDelegationTokenProvider.scala | 3 + .../security/HadoopFSDelegationTokenProvider.scala | 8 +- ...k.deploy.security.HadoopDelegationTokenProvider | 1 + .../HadoopDelegationTokenManagerSuite.scala | 102 ++++++--------------- external/kafka-0-10-sql/pom.xml | 5 + .../spark/sql/kafka010/KafkaConfigUpdater.scala | 2 +- .../spark/sql/kafka010/KafkaSecurityHelper.scala | 2 +- .../sql/kafka010/KafkaDelegationTokenTest.scala | 4 +- .../pom.xml | 72 +-------------- ...k.deploy.security.HadoopDelegationTokenProvider | 1 + .../kafka010}/KafkaDelegationTokenProvider.scala | 9 +- .../apache/spark/kafka010}/KafkaTokenUtil.scala | 10 +- .../src/test/resources/log4j.properties | 28 ++++++ .../KafkaHadoopDelegationTokenManagerSuite.scala | 32 +++++++ .../spark/kafka010}/KafkaTokenUtilSuite.scala | 2 +- pom.xml | 1 + project/MimaExcludes.scala | 3 + project/SparkBuild.scala | 6 +- .../sql/hive/thriftserver/SparkSQLCLIDriver.scala | 4 +- ...k.deploy.security.HadoopDelegationTokenProvider | 1 + .../security/HiveDelegationTokenProvider.scala | 5 +- .../HiveHadoopDelegationTokenManagerSuite.scala | 23 +---- 26 files changed, 165 insertions(+), 241 deletions(-) diff --git a/core/pom.xml b/core/pom.xml index 1cd1ad9..c87d9d5 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -385,46 +385,6 @@ <artifactId>commons-crypto</artifactId> </dependency> - <!-- - The following dependencies are depended upon in HiveCredentialProvider, but are only executed if Hive is enabled in - the user's Hadoop configuration. So in order to prevent spark-core from depending on Hive, these deps have been - placed in the "provided" scope, rather than the "compile" scope, and NoClassDefFoundError exceptions are handled - when the user has not explicitly compiled with the Hive module. - --> - <dependency> - <groupId>${hive.group}</groupId> - <artifactId>hive-exec</artifactId> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>${hive.group}</groupId> - <artifactId>hive-metastore</artifactId> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.thrift</groupId> - <artifactId>libthrift</artifactId> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.thrift</groupId> - <artifactId>libfb303</artifactId> - <scope>provided</scope> - </dependency> - - <!-- - The following kafka dependency used to obtain delegation token. - In order to prevent spark-core from depending on kafka, these deps have been placed in the - "provided" scope, rather than the "compile" scope, and NoClassDefFoundError exceptions are - handled when the user explicitly use neither spark-streaming-kafka nor spark-sql-kafka modules. - --> - <dependency> - <groupId>org.apache.kafka</groupId> - <artifactId>kafka-clients</artifactId> - <version>${kafka.version}</version> - <scope>provided</scope> - </dependency> - </dependencies> <build> <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory> diff --git a/core/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider b/core/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider new file mode 100644 index 0000000..c1f2060 --- /dev/null +++ b/core/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider @@ -0,0 +1,2 @@ +org.apache.spark.deploy.security.HadoopFSDelegationTokenProvider +org.apache.spark.deploy.security.HBaseDelegationTokenProvider diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala b/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala index 6ef6835..3bf8c14 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala @@ -21,6 +21,7 @@ import scala.reflect.runtime.universe import scala.util.control.NonFatal import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.security.Credentials import org.apache.hadoop.security.token.{Token, TokenIdentifier} @@ -36,6 +37,7 @@ private[security] class HBaseDelegationTokenProvider override def obtainDelegationTokens( hadoopConf: Configuration, sparkConf: SparkConf, + fileSystems: Set[FileSystem], creds: Credentials): Option[Long] = { try { val mirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader) diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala index d97857a..2763a46 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala @@ -20,8 +20,11 @@ package org.apache.spark.deploy.security import java.io.File import java.net.URI import java.security.PrivilegedExceptionAction +import java.util.ServiceLoader import java.util.concurrent.{ScheduledExecutorService, TimeUnit} +import scala.collection.mutable + import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.security.{Credentials, UserGroupInformation} @@ -33,7 +36,7 @@ import org.apache.spark.internal.config._ import org.apache.spark.rpc.RpcEndpointRef import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens import org.apache.spark.ui.UIUtils -import org.apache.spark.util.ThreadUtils +import org.apache.spark.util.{ThreadUtils, Utils} /** * Manager for delegation tokens in a Spark application. @@ -132,7 +135,7 @@ private[spark] class HadoopDelegationTokenManager( def obtainDelegationTokens(creds: Credentials): Long = { delegationTokenProviders.values.flatMap { provider => if (provider.delegationTokensRequired(sparkConf, hadoopConf)) { - provider.obtainDelegationTokens(hadoopConf, sparkConf, creds) + provider.obtainDelegationTokens(hadoopConf, sparkConf, fileSystemsToAccess(), creds) } else { logDebug(s"Service ${provider.serviceName} does not require a token." + s" Check your configuration to see if security is disabled or not.") @@ -244,12 +247,19 @@ private[spark] class HadoopDelegationTokenManager( } private def loadProviders(): Map[String, HadoopDelegationTokenProvider] = { - val providers = Seq( - new HadoopFSDelegationTokenProvider( - () => HadoopDelegationTokenManager.this.fileSystemsToAccess())) ++ - safeCreateProvider(new HiveDelegationTokenProvider) ++ - safeCreateProvider(new HBaseDelegationTokenProvider) ++ - safeCreateProvider(new KafkaDelegationTokenProvider) + val loader = ServiceLoader.load(classOf[HadoopDelegationTokenProvider], + Utils.getContextOrSparkClassLoader) + val providers = mutable.ArrayBuffer[HadoopDelegationTokenProvider]() + + val iterator = loader.iterator + while (iterator.hasNext) { + try { + providers += iterator.next + } catch { + case t: Throwable => + logDebug(s"Failed to load built in provider.", t) + } + } // Filter out providers for which spark.security.credentials.{service}.enabled is false. providers @@ -257,16 +267,4 @@ private[spark] class HadoopDelegationTokenManager( .map { p => (p.serviceName, p) } .toMap } - - private def safeCreateProvider( - createFn: => HadoopDelegationTokenProvider): Option[HadoopDelegationTokenProvider] = { - try { - Some(createFn) - } catch { - case t: Throwable => - logDebug(s"Failed to load built in provider.", t) - None - } - } - } diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenProvider.scala b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenProvider.scala index ed09050..cb4c97b 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenProvider.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenProvider.scala @@ -18,6 +18,7 @@ package org.apache.spark.deploy.security import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.security.Credentials import org.apache.spark.SparkConf @@ -43,11 +44,13 @@ private[spark] trait HadoopDelegationTokenProvider { * Obtain delegation tokens for this service and get the time of the next renewal. * @param hadoopConf Configuration of current Hadoop Compatible system. * @param creds Credentials to add tokens and security keys to. + * @param fileSystems List of file systems for which to obtain delegation tokens. * @return If the returned tokens are renewable and can be renewed, return the time of the next * renewal, otherwise None should be returned. */ def obtainDelegationTokens( hadoopConf: Configuration, sparkConf: SparkConf, + fileSystems: Set[FileSystem], creds: Credentials): Option[Long] } diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala b/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala index 00200f8..2cd160c 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala @@ -31,7 +31,7 @@ import org.apache.spark.{SparkConf, SparkException} import org.apache.spark.internal.Logging import org.apache.spark.internal.config._ -private[deploy] class HadoopFSDelegationTokenProvider(fileSystems: () => Set[FileSystem]) +private[deploy] class HadoopFSDelegationTokenProvider extends HadoopDelegationTokenProvider with Logging { // This tokenRenewalInterval will be set in the first call to obtainDelegationTokens. @@ -44,14 +44,14 @@ private[deploy] class HadoopFSDelegationTokenProvider(fileSystems: () => Set[Fil override def obtainDelegationTokens( hadoopConf: Configuration, sparkConf: SparkConf, + fileSystems: Set[FileSystem], creds: Credentials): Option[Long] = { try { - val fsToGetTokens = fileSystems() - val fetchCreds = fetchDelegationTokens(getTokenRenewer(hadoopConf), fsToGetTokens, creds) + val fetchCreds = fetchDelegationTokens(getTokenRenewer(hadoopConf), fileSystems, creds) // Get the token renewal interval if it is not set. It will only be called once. if (tokenRenewalInterval == null) { - tokenRenewalInterval = getTokenRenewalInterval(hadoopConf, sparkConf, fsToGetTokens) + tokenRenewalInterval = getTokenRenewalInterval(hadoopConf, sparkConf, fileSystems) } // Get the time of next renewal. diff --git a/core/src/test/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider b/core/src/test/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider new file mode 100644 index 0000000..f4107be --- /dev/null +++ b/core/src/test/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider @@ -0,0 +1 @@ +org.apache.spark.deploy.security.ExceptionThrowingDelegationTokenProvider diff --git a/core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala index af7d44b..9cc4a91 100644 --- a/core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala @@ -17,99 +17,57 @@ package org.apache.spark.deploy.security -import org.apache.commons.io.IOUtils import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.FileSystem +import org.apache.hadoop.security.Credentials import org.apache.spark.{SparkConf, SparkFunSuite} -import org.apache.spark.util.Utils + +private class ExceptionThrowingDelegationTokenProvider extends HadoopDelegationTokenProvider { + ExceptionThrowingDelegationTokenProvider.constructed = true + throw new IllegalArgumentException + + override def serviceName: String = "throw" + + override def delegationTokensRequired( + sparkConf: SparkConf, + hadoopConf: Configuration): Boolean = throw new IllegalArgumentException + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + fileSystems: Set[FileSystem], + creds: Credentials): Option[Long] = throw new IllegalArgumentException +} + +private object ExceptionThrowingDelegationTokenProvider { + var constructed = false +} class HadoopDelegationTokenManagerSuite extends SparkFunSuite { private val hadoopConf = new Configuration() test("default configuration") { + ExceptionThrowingDelegationTokenProvider.constructed = false val manager = new HadoopDelegationTokenManager(new SparkConf(false), hadoopConf, null) assert(manager.isProviderLoaded("hadoopfs")) assert(manager.isProviderLoaded("hbase")) - assert(manager.isProviderLoaded("hive")) - assert(manager.isProviderLoaded("kafka")) + // This checks that providers are loaded independently and they have no effect on each other + assert(ExceptionThrowingDelegationTokenProvider.constructed) + assert(!manager.isProviderLoaded("throw")) } - test("disable hive credential provider") { - val sparkConf = new SparkConf(false).set("spark.security.credentials.hive.enabled", "false") + test("disable hadoopfs credential provider") { + val sparkConf = new SparkConf(false).set("spark.security.credentials.hadoopfs.enabled", "false") val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) - assert(manager.isProviderLoaded("hadoopfs")) - assert(manager.isProviderLoaded("hbase")) - assert(!manager.isProviderLoaded("hive")) - assert(manager.isProviderLoaded("kafka")) + assert(!manager.isProviderLoaded("hadoopfs")) } test("using deprecated configurations") { val sparkConf = new SparkConf(false) .set("spark.yarn.security.tokens.hadoopfs.enabled", "false") - .set("spark.yarn.security.credentials.hive.enabled", "false") val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) assert(!manager.isProviderLoaded("hadoopfs")) assert(manager.isProviderLoaded("hbase")) - assert(!manager.isProviderLoaded("hive")) - assert(manager.isProviderLoaded("kafka")) - } - - test("SPARK-23209: obtain tokens when Hive classes are not available") { - // This test needs a custom class loader to hide Hive classes which are in the classpath. - // Because the manager code loads the Hive provider directly instead of using reflection, we - // need to drive the test through the custom class loader so a new copy that cannot find - // Hive classes is loaded. - val currentLoader = Thread.currentThread().getContextClassLoader() - val noHive = new ClassLoader() { - override def loadClass(name: String, resolve: Boolean): Class[_] = { - if (name.startsWith("org.apache.hive") || name.startsWith("org.apache.hadoop.hive")) { - throw new ClassNotFoundException(name) - } - - val prefixBlacklist = Seq("java", "scala", "com.sun.", "sun.") - if (prefixBlacklist.exists(name.startsWith(_))) { - return currentLoader.loadClass(name) - } - - val found = findLoadedClass(name) - if (found != null) { - return found - } - - val classFileName = name.replaceAll("\\.", "/") + ".class" - val in = currentLoader.getResourceAsStream(classFileName) - if (in != null) { - val bytes = IOUtils.toByteArray(in) - return defineClass(name, bytes, 0, bytes.length) - } - - throw new ClassNotFoundException(name) - } - } - - Utils.withContextClassLoader(noHive) { - val test = noHive.loadClass(NoHiveTest.getClass.getName().stripSuffix("$")) - test.getMethod("runTest").invoke(null) - } } } - -/** Test code for SPARK-23209 to avoid using too much reflection above. */ -private object NoHiveTest { - - def runTest(): Unit = { - try { - val manager = new HadoopDelegationTokenManager(new SparkConf(), new Configuration(), null) - require(!manager.isProviderLoaded("hive")) - } catch { - case e: Throwable => - // Throw a better exception in case the test fails, since there may be a lot of nesting. - var cause = e - while (cause.getCause() != null) { - cause = cause.getCause() - } - throw cause - } - } - -} diff --git a/external/kafka-0-10-sql/pom.xml b/external/kafka-0-10-sql/pom.xml index 1c77906..827ceb8 100644 --- a/external/kafka-0-10-sql/pom.xml +++ b/external/kafka-0-10-sql/pom.xml @@ -37,6 +37,11 @@ <dependencies> <dependency> <groupId>org.apache.spark</groupId> + <artifactId>spark-token-provider-kafka-0-10_${scala.binary.version}</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.spark</groupId> <artifactId>spark-sql_${scala.binary.version}</artifactId> <version>${project.version}</version> <scope>provided</scope> diff --git a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaConfigUpdater.scala b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaConfigUpdater.scala index bc1b801..38bf5d7 100644 --- a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaConfigUpdater.scala +++ b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaConfigUpdater.scala @@ -24,9 +24,9 @@ import scala.collection.JavaConverters._ import org.apache.kafka.common.config.SaslConfigs import org.apache.spark.SparkEnv -import org.apache.spark.deploy.security.KafkaTokenUtil import org.apache.spark.internal.Logging import org.apache.spark.internal.config.Kafka +import org.apache.spark.kafka010.KafkaTokenUtil /** * Class to conveniently update Kafka config params, while logging the changes diff --git a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSecurityHelper.scala b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSecurityHelper.scala index 7215295..a11d54f 100644 --- a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSecurityHelper.scala +++ b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSecurityHelper.scala @@ -21,9 +21,9 @@ import org.apache.hadoop.security.UserGroupInformation import org.apache.kafka.common.security.scram.ScramLoginModule import org.apache.spark.SparkConf -import org.apache.spark.deploy.security.KafkaTokenUtil import org.apache.spark.internal.Logging import org.apache.spark.internal.config._ +import org.apache.spark.kafka010.KafkaTokenUtil private[kafka010] object KafkaSecurityHelper extends Logging { def isTokenAvailable(): Boolean = { diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenTest.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenTest.scala index 31247ab..d0cefc4 100644 --- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenTest.scala +++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenTest.scala @@ -26,8 +26,8 @@ import org.mockito.Mockito.mock import org.scalatest.BeforeAndAfterEach import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite} -import org.apache.spark.deploy.security.KafkaTokenUtil -import org.apache.spark.deploy.security.KafkaTokenUtil.KafkaDelegationTokenIdentifier +import org.apache.spark.kafka010.KafkaTokenUtil +import org.apache.spark.kafka010.KafkaTokenUtil.KafkaDelegationTokenIdentifier /** * This is a trait which provides functionalities for Kafka delegation token related test suites. diff --git a/external/kafka-0-10-sql/pom.xml b/external/kafka-0-10-token-provider/pom.xml similarity index 53% copy from external/kafka-0-10-sql/pom.xml copy to external/kafka-0-10-token-provider/pom.xml index 1c77906..b2abcd9 100644 --- a/external/kafka-0-10-sql/pom.xml +++ b/external/kafka-0-10-token-provider/pom.xml @@ -26,38 +26,23 @@ </parent> <groupId>org.apache.spark</groupId> - <artifactId>spark-sql-kafka-0-10_2.12</artifactId> + <artifactId>spark-token-provider-kafka-0-10_2.12</artifactId> <properties> - <sbt.project.name>sql-kafka-0-10</sbt.project.name> + <sbt.project.name>token-provider-kafka-0-10</sbt.project.name> </properties> <packaging>jar</packaging> - <name>Kafka 0.10+ Source for Structured Streaming</name> + <name>Kafka 0.10+ Token Provider for Streaming</name> <url>http://spark.apache.org/</url> <dependencies> <dependency> <groupId>org.apache.spark</groupId> - <artifactId>spark-sql_${scala.binary.version}</artifactId> - <version>${project.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.spark</groupId> <artifactId>spark-core_${scala.binary.version}</artifactId> <version>${project.version}</version> - <type>test-jar</type> - <scope>test</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> - <artifactId>spark-catalyst_${scala.binary.version}</artifactId> - <version>${project.version}</version> - <type>test-jar</type> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.apache.spark</groupId> - <artifactId>spark-sql_${scala.binary.version}</artifactId> + <artifactId>spark-core_${scala.binary.version}</artifactId> <version>${project.version}</version> <type>test-jar</type> <scope>test</scope> @@ -68,55 +53,6 @@ <version>${kafka.version}</version> </dependency> <dependency> - <groupId>org.apache.kafka</groupId> - <artifactId>kafka_${scala.binary.version}</artifactId> - <version>${kafka.version}</version> - <scope>test</scope> - <exclusions> - <exclusion> - <groupId>com.fasterxml.jackson.core</groupId> - <artifactId>jackson-core</artifactId> - </exclusion> - <exclusion> - <groupId>com.fasterxml.jackson.core</groupId> - <artifactId>jackson-databind</artifactId> - </exclusion> - <exclusion> - <groupId>com.fasterxml.jackson.core</groupId> - <artifactId>jackson-annotations</artifactId> - </exclusion> - </exclusions> - </dependency> - <!-- Kafka embedded server uses Zookeeper 3.4.7 API --> - <dependency> - <groupId>org.apache.zookeeper</groupId> - <artifactId>zookeeper</artifactId> - <version>3.4.7</version> - <scope>test</scope> - </dependency> - <dependency> - <groupId>net.sf.jopt-simple</groupId> - <artifactId>jopt-simple</artifactId> - <version>3.2</version> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.eclipse.jetty</groupId> - <artifactId>jetty-servlet</artifactId> - <version>${jetty.version}</version> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.mockito</groupId> - <artifactId>mockito-core</artifactId> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.scalacheck</groupId> - <artifactId>scalacheck_${scala.binary.version}</artifactId> - <scope>test</scope> - </dependency> - <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-tags_${scala.binary.version}</artifactId> </dependency> diff --git a/external/kafka-0-10-token-provider/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider b/external/kafka-0-10-token-provider/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider new file mode 100644 index 0000000..3401401 --- /dev/null +++ b/external/kafka-0-10-token-provider/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider @@ -0,0 +1 @@ +org.apache.spark.kafka010.KafkaDelegationTokenProvider diff --git a/core/src/main/scala/org/apache/spark/deploy/security/KafkaDelegationTokenProvider.scala b/external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaDelegationTokenProvider.scala similarity index 88% rename from core/src/main/scala/org/apache/spark/deploy/security/KafkaDelegationTokenProvider.scala rename to external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaDelegationTokenProvider.scala index f67cb26..ac6baa0 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/KafkaDelegationTokenProvider.scala +++ b/external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaDelegationTokenProvider.scala @@ -15,20 +15,22 @@ * limitations under the License. */ -package org.apache.spark.deploy.security +package org.apache.spark.kafka010 import scala.language.existentials import scala.util.control.NonFatal import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.security.Credentials import org.apache.kafka.common.security.auth.SecurityProtocol.{SASL_PLAINTEXT, SASL_SSL, SSL} import org.apache.spark.SparkConf +import org.apache.spark.deploy.security.HadoopDelegationTokenProvider import org.apache.spark.internal.Logging -import org.apache.spark.internal.config._ +import org.apache.spark.internal.config.Kafka -private[security] class KafkaDelegationTokenProvider +private[spark] class KafkaDelegationTokenProvider extends HadoopDelegationTokenProvider with Logging { override def serviceName: String = "kafka" @@ -36,6 +38,7 @@ private[security] class KafkaDelegationTokenProvider override def obtainDelegationTokens( hadoopConf: Configuration, sparkConf: SparkConf, + fileSystems: Set[FileSystem], creds: Credentials): Option[Long] = { try { logDebug("Attempting to fetch Kafka security token.") diff --git a/core/src/main/scala/org/apache/spark/deploy/security/KafkaTokenUtil.scala b/external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala similarity index 96% rename from core/src/main/scala/org/apache/spark/deploy/security/KafkaTokenUtil.scala rename to external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala index f363853..574d58b 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/KafkaTokenUtil.scala +++ b/external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.spark.deploy.security +package org.apache.spark.kafka010 import java.{util => ju} import java.text.SimpleDateFormat @@ -46,7 +46,7 @@ private[spark] object KafkaTokenUtil extends Logging { override def getKind: Text = TOKEN_KIND } - private[security] def obtainToken(sparkConf: SparkConf): (Token[_ <: TokenIdentifier], Long) = { + private[kafka010] def obtainToken(sparkConf: SparkConf): (Token[_ <: TokenIdentifier], Long) = { checkProxyUser() val adminClient = AdminClient.create(createAdminClientProperties(sparkConf)) @@ -63,7 +63,7 @@ private[spark] object KafkaTokenUtil extends Logging { ), token.tokenInfo.expiryTimestamp) } - private[security] def checkProxyUser(): Unit = { + private[kafka010] def checkProxyUser(): Unit = { val currentUser = UserGroupInformation.getCurrentUser() // Obtaining delegation token for proxy user is planned but not yet implemented // See https://issues.apache.org/jira/browse/KAFKA-6945 @@ -71,7 +71,7 @@ private[spark] object KafkaTokenUtil extends Logging { "user is not yet supported.") } - private[security] def createAdminClientProperties(sparkConf: SparkConf): ju.Properties = { + private[kafka010] def createAdminClientProperties(sparkConf: SparkConf): ju.Properties = { val adminClientProperties = new ju.Properties val bootstrapServers = sparkConf.get(Kafka.BOOTSTRAP_SERVERS) @@ -154,7 +154,7 @@ private[spark] object KafkaTokenUtil extends Logging { } } - private[security] def getKeytabJaasParams(sparkConf: SparkConf): String = { + private[kafka010] def getKeytabJaasParams(sparkConf: SparkConf): String = { val params = s""" |${getKrb5LoginModuleName} required diff --git a/external/kafka-0-10-token-provider/src/test/resources/log4j.properties b/external/kafka-0-10-token-provider/src/test/resources/log4j.properties new file mode 100644 index 0000000..75e3b53 --- /dev/null +++ b/external/kafka-0-10-token-provider/src/test/resources/log4j.properties @@ -0,0 +1,28 @@ +# +# 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. +# + +# Set everything to be logged to the file target/unit-tests.log +log4j.rootCategory=INFO, file +log4j.appender.file=org.apache.log4j.FileAppender +log4j.appender.file.append=true +log4j.appender.file.file=target/unit-tests.log +log4j.appender.file.layout=org.apache.log4j.PatternLayout +log4j.appender.file.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss.SSS} %t %p %c{1}: %m%n + +# Ignore messages below warning level from Jetty, because it's a bit verbose +log4j.logger.org.spark-project.jetty=WARN + diff --git a/external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaHadoopDelegationTokenManagerSuite.scala b/external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaHadoopDelegationTokenManagerSuite.scala new file mode 100644 index 0000000..9aa7618 --- /dev/null +++ b/external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaHadoopDelegationTokenManagerSuite.scala @@ -0,0 +1,32 @@ +/* + * 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.kafka010 + +import org.apache.hadoop.conf.Configuration + +import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.deploy.security.HadoopDelegationTokenManager + +class KafkaHadoopDelegationTokenManagerSuite extends SparkFunSuite { + private val hadoopConf = new Configuration() + + test("default configuration") { + val manager = new HadoopDelegationTokenManager(new SparkConf(false), hadoopConf, null) + assert(manager.isProviderLoaded("kafka")) + } +} diff --git a/core/src/test/scala/org/apache/spark/deploy/security/KafkaTokenUtilSuite.scala b/external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala similarity index 99% rename from core/src/test/scala/org/apache/spark/deploy/security/KafkaTokenUtilSuite.scala rename to external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala index daa7e54..5da6260 100644 --- a/core/src/test/scala/org/apache/spark/deploy/security/KafkaTokenUtilSuite.scala +++ b/external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.spark.deploy.security +package org.apache.spark.kafka010 import java.{util => ju} import java.security.PrivilegedExceptionAction diff --git a/pom.xml b/pom.xml index de14d6a..29a281a 100644 --- a/pom.xml +++ b/pom.xml @@ -101,6 +101,7 @@ <module>examples</module> <module>repl</module> <module>launcher</module> + <module>external/kafka-0-10-token-provider</module> <module>external/kafka-0-10</module> <module>external/kafka-0-10-assembly</module> <module>external/kafka-0-10-sql</module> diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index 4cf312d..0cdef00 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -36,6 +36,9 @@ object MimaExcludes { // Exclude rules for 3.0.x lazy val v30excludes = v24excludes ++ Seq( + // [SPARK-26254][CORE] Extract Hive + Kafka dependencies from Core. + ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.deploy.security.HiveDelegationTokenProvider"), + // [SPARK-25765][ML] Add training cost to BisectingKMeans summary ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.mllib.clustering.BisectingKMeansModel.this"), diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala index a8f0ca4..8d836da 100644 --- a/project/SparkBuild.scala +++ b/project/SparkBuild.scala @@ -41,8 +41,8 @@ object BuildCommons { private val buildLocation = file(".").getAbsoluteFile.getParentFile - val sqlProjects@Seq(catalyst, sql, hive, hiveThriftServer, sqlKafka010, avro) = Seq( - "catalyst", "sql", "hive", "hive-thriftserver", "sql-kafka-0-10", "avro" + val sqlProjects@Seq(catalyst, sql, hive, hiveThriftServer, tokenProviderKafka010, sqlKafka010, avro) = Seq( + "catalyst", "sql", "hive", "hive-thriftserver", "token-provider-kafka-0-10", "sql-kafka-0-10", "avro" ).map(ProjectRef(buildLocation, _)) val streamingProjects@Seq(streaming, streamingKafka010) = @@ -331,7 +331,7 @@ object SparkBuild extends PomBuild { val mimaProjects = allProjects.filterNot { x => Seq( spark, hive, hiveThriftServer, catalyst, repl, networkCommon, networkShuffle, networkYarn, - unsafe, tags, sqlKafka010, kvstore, avro + unsafe, tags, tokenProviderKafka010, sqlKafka010, kvstore, avro ).contains(x) } diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala index bb96cea..463cf16 100644 --- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala +++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLCLIDriver.scala @@ -40,10 +40,10 @@ import org.apache.thrift.transport.TSocket import org.apache.spark.SparkConf import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.HiveDelegationTokenProvider import org.apache.spark.internal.Logging import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.hive.HiveUtils +import org.apache.spark.sql.hive.security.HiveDelegationTokenProvider import org.apache.spark.util.ShutdownHookManager /** @@ -126,7 +126,7 @@ private[hive] object SparkSQLCLIDriver extends Logging { val tokenProvider = new HiveDelegationTokenProvider() if (tokenProvider.delegationTokensRequired(sparkConf, hadoopConf)) { val credentials = new Credentials() - tokenProvider.obtainDelegationTokens(hadoopConf, sparkConf, credentials) + tokenProvider.obtainDelegationTokens(hadoopConf, sparkConf, Set.empty, credentials) UserGroupInformation.getCurrentUser.addCredentials(credentials) } diff --git a/sql/hive/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider b/sql/hive/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider new file mode 100644 index 0000000..2b0acc0 --- /dev/null +++ b/sql/hive/src/main/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider @@ -0,0 +1 @@ +org.apache.spark.sql.hive.security.HiveDelegationTokenProvider diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HiveDelegationTokenProvider.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/security/HiveDelegationTokenProvider.scala similarity index 96% rename from core/src/main/scala/org/apache/spark/deploy/security/HiveDelegationTokenProvider.scala rename to sql/hive/src/main/scala/org/apache/spark/sql/hive/security/HiveDelegationTokenProvider.scala index 4ca0136..25eb251 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/HiveDelegationTokenProvider.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/security/HiveDelegationTokenProvider.scala @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.spark.deploy.security +package org.apache.spark.sql.hive.security import java.lang.reflect.UndeclaredThrowableException import java.security.PrivilegedExceptionAction @@ -23,6 +23,7 @@ import java.security.PrivilegedExceptionAction import scala.util.control.NonFatal import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier import org.apache.hadoop.hive.conf.HiveConf import org.apache.hadoop.hive.ql.metadata.Hive @@ -32,6 +33,7 @@ import org.apache.hadoop.security.token.Token import org.apache.spark.SparkConf import org.apache.spark.deploy.SparkHadoopUtil +import org.apache.spark.deploy.security.HadoopDelegationTokenProvider import org.apache.spark.internal.Logging import org.apache.spark.internal.config.KEYTAB import org.apache.spark.util.Utils @@ -83,6 +85,7 @@ private[spark] class HiveDelegationTokenProvider override def obtainDelegationTokens( hadoopConf: Configuration, sparkConf: SparkConf, + fileSystems: Set[FileSystem], creds: Credentials): Option[Long] = { try { val conf = hiveConf(hadoopConf) diff --git a/core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/security/HiveHadoopDelegationTokenManagerSuite.scala similarity index 80% copy from core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala copy to sql/hive/src/test/scala/org/apache/spark/sql/hive/security/HiveHadoopDelegationTokenManagerSuite.scala index af7d44b..ce40cf5 100644 --- a/core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/security/HiveHadoopDelegationTokenManagerSuite.scala @@ -15,43 +15,28 @@ * limitations under the License. */ -package org.apache.spark.deploy.security +package org.apache.spark.sql.hive.security import org.apache.commons.io.IOUtils import org.apache.hadoop.conf.Configuration import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.deploy.security.HadoopDelegationTokenManager import org.apache.spark.util.Utils -class HadoopDelegationTokenManagerSuite extends SparkFunSuite { +class HiveHadoopDelegationTokenManagerSuite extends SparkFunSuite { private val hadoopConf = new Configuration() test("default configuration") { val manager = new HadoopDelegationTokenManager(new SparkConf(false), hadoopConf, null) - assert(manager.isProviderLoaded("hadoopfs")) - assert(manager.isProviderLoaded("hbase")) assert(manager.isProviderLoaded("hive")) - assert(manager.isProviderLoaded("kafka")) - } - - test("disable hive credential provider") { - val sparkConf = new SparkConf(false).set("spark.security.credentials.hive.enabled", "false") - val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) - assert(manager.isProviderLoaded("hadoopfs")) - assert(manager.isProviderLoaded("hbase")) - assert(!manager.isProviderLoaded("hive")) - assert(manager.isProviderLoaded("kafka")) } test("using deprecated configurations") { val sparkConf = new SparkConf(false) - .set("spark.yarn.security.tokens.hadoopfs.enabled", "false") .set("spark.yarn.security.credentials.hive.enabled", "false") val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) - assert(!manager.isProviderLoaded("hadoopfs")) - assert(manager.isProviderLoaded("hbase")) assert(!manager.isProviderLoaded("hive")) - assert(manager.isProviderLoaded("kafka")) } test("SPARK-23209: obtain tokens when Hive classes are not available") { @@ -100,6 +85,8 @@ private object NoHiveTest { def runTest(): Unit = { try { val manager = new HadoopDelegationTokenManager(new SparkConf(), new Configuration(), null) + assert(manager.isProviderLoaded("hadoopfs")) + assert(manager.isProviderLoaded("hbase")) require(!manager.isProviderLoaded("hive")) } catch { case e: Throwable => --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org