From 5d81e21b2148133a165768016a91ea69208b6981 Mon Sep 17 00:00:00 2001 From: zhou-jiang Date: Mon, 4 Nov 2019 16:14:12 -0800 Subject: [PATCH 01/10] [SPARK-25694]URL.setURLStreamHandlerFactory causing incompatible HttpURLConnection issue --- .../spark/sql/internal/SharedState.scala | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index b810bedac471d..e8f1d2df21ea1 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -52,6 +52,8 @@ private[sql] class SharedState( initialConfigs: scala.collection.Map[String, String]) extends Logging { + SharedState.setFsUrlStreamHandlerFactoryIfNeeded(sparkContext.conf) + // Load hive-site.xml into hadoopConf and determine the warehouse path we want to use, based on // the config from both hive and Spark SQL. Finally set the warehouse config value to sparkConf. val warehousePath: String = { @@ -191,11 +193,24 @@ private[sql] class SharedState( } object SharedState extends Logging { - try { - URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) - } catch { - case e: Error => - logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") + private var initialized = false + private def setFsUrlStreamHandlerFactoryIfNeeded(conf: SparkConf): Unit = { + synchronized { + if (!initialized) { + try { + if (conf.getBoolean("spark.fsUrlStreamHandlerFactory.enabled", true)) { + URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) + } + } catch { + case e: Error => + logWarning("URL.setURLStreamHandlerFactory failed to set " + + "FsUrlStreamHandlerFactory", e) + } finally { + // don't retry on failure + initialized = true + } + } + } } private val HIVE_EXTERNAL_CATALOG_CLASS_NAME = "org.apache.spark.sql.hive.HiveExternalCatalog" From 1e4d5c16d1ee37806bcca145eaceb543668e7757 Mon Sep 17 00:00:00 2001 From: zhou-jiang Date: Thu, 14 Nov 2019 10:59:09 -0800 Subject: [PATCH 02/10] Add double check locking for UrlStreamHandlerFactory. --- .../spark/sql/internal/SharedState.scala | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index e8f1d2df21ea1..961829e65c179 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -52,7 +52,7 @@ private[sql] class SharedState( initialConfigs: scala.collection.Map[String, String]) extends Logging { - SharedState.setFsUrlStreamHandlerFactoryIfNeeded(sparkContext.conf) + SharedState.setFsUrlStreamHandlerFactory(sparkContext.conf) // Load hive-site.xml into hadoopConf and determine the warehouse path we want to use, based on // the config from both hive and Spark SQL. Finally set the warehouse config value to sparkConf. @@ -193,21 +193,23 @@ private[sql] class SharedState( } object SharedState extends Logging { - private var initialized = false - private def setFsUrlStreamHandlerFactoryIfNeeded(conf: SparkConf): Unit = { - synchronized { - if (!initialized) { + @volatile private var factory: Option[FsUrlStreamHandlerFactory] = None + private lazy val defaultFactory = new FsUrlStreamHandlerFactory() + private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { + factory match { + case Some(_) => + logWarning("FsUrlStreamHandlerFactory has been already initialized, " + + "so it can not be modified") + case None => synchronized { try { if (conf.getBoolean("spark.fsUrlStreamHandlerFactory.enabled", true)) { - URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) + URL.setURLStreamHandlerFactory(defaultFactory) + factory = Some(defaultFactory) } } catch { case e: Error => logWarning("URL.setURLStreamHandlerFactory failed to set " + "FsUrlStreamHandlerFactory", e) - } finally { - // don't retry on failure - initialized = true } } } From 1970254e41f0ec017d59d7e896978fde8bb07603 Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 13:29:25 -0800 Subject: [PATCH 03/10] Address comments --- .../spark/sql/internal/SharedState.scala | 3 +- .../spark/sql/internal/config/package.scala | 29 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/internal/config/package.scala diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index 961829e65c179..345027d8b14c8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -36,6 +36,7 @@ import org.apache.spark.sql.execution.CacheManager import org.apache.spark.sql.execution.streaming.StreamExecution import org.apache.spark.sql.execution.ui.{SQLAppStatusListener, SQLAppStatusStore, SQLTab} import org.apache.spark.sql.internal.StaticSQLConf._ +import org.apache.spark.sql.internal.config.DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED import org.apache.spark.sql.streaming.StreamingQuery import org.apache.spark.status.ElementTrackingStore import org.apache.spark.util.Utils @@ -202,7 +203,7 @@ object SharedState extends Logging { "so it can not be modified") case None => synchronized { try { - if (conf.getBoolean("spark.fsUrlStreamHandlerFactory.enabled", true)) { + if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED)) { URL.setURLStreamHandlerFactory(defaultFactory) factory = Some(defaultFactory) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/config/package.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/config/package.scala new file mode 100644 index 0000000000000..e26c4aadaf135 --- /dev/null +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/config/package.scala @@ -0,0 +1,29 @@ +/* + * 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.sql.internal + +import org.apache.spark.internal.config.ConfigBuilder + +package object config { + + private[spark] val DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED = + ConfigBuilder("spark.sql.defaultUrlStreamHandlerFactory.enabled") + .doc("When true, set FsUrlStreamHandlerFactory to support ADD JAR against HDFS locations") + .booleanConf + .createWithDefault(true) +} From 4367b1590bde2dd9fc4fa286cac7ffc6ba10b00d Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 15:37:23 -0800 Subject: [PATCH 04/10] simplify --- .../spark/sql/internal/SharedState.scala | 23 +++++-------------- 1 file changed, 6 insertions(+), 17 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index 345027d8b14c8..004447009d5b0 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -194,24 +194,13 @@ private[sql] class SharedState( } object SharedState extends Logging { - @volatile private var factory: Option[FsUrlStreamHandlerFactory] = None - private lazy val defaultFactory = new FsUrlStreamHandlerFactory() private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { - factory match { - case Some(_) => - logWarning("FsUrlStreamHandlerFactory has been already initialized, " + - "so it can not be modified") - case None => synchronized { - try { - if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED)) { - URL.setURLStreamHandlerFactory(defaultFactory) - factory = Some(defaultFactory) - } - } catch { - case e: Error => - logWarning("URL.setURLStreamHandlerFactory failed to set " + - "FsUrlStreamHandlerFactory", e) - } + if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED)) { + try { + URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) + } catch { + case _: Error => + logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") } } } From 558062b43a6e758976d2eb1c7f4fa48ef2a3b76c Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 15:49:13 -0800 Subject: [PATCH 05/10] Use NonFatal --- .../scala/org/apache/spark/sql/internal/SharedState.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index 004447009d5b0..7782400f67ba8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -199,8 +199,8 @@ object SharedState extends Logging { try { URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) } catch { - case _: Error => - logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") + case NonFatal(e) => + logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory", e) } } } From 346ae61ddf30b1afd64266ce6be17c1e9f55b469 Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 17:19:23 -0800 Subject: [PATCH 06/10] Address comments --- .../spark/sql/internal/SharedState.scala | 22 ++++++++++++++----- 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index 7782400f67ba8..c82d6a781983f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -195,16 +195,26 @@ private[sql] class SharedState( object SharedState extends Logging { private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { - if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED)) { - try { - URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) - } catch { - case NonFatal(e) => - logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory", e) + if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED) && factory.isEmpty) { + factory.synchronized { + if (factory.isEmpty) { + try { + URL.setURLStreamHandlerFactory(defaultFactory) + factory = Some(defaultFactory) + } catch { + case NonFatal(e) => + logWarning("URL.setURLStreamHandlerFactory failed to set " + + "FsUrlStreamHandlerFactory", e) + } + } } } } + @volatile private var factory: Option[FsUrlStreamHandlerFactory] = None + + private lazy val defaultFactory = new FsUrlStreamHandlerFactory() + private val HIVE_EXTERNAL_CATALOG_CLASS_NAME = "org.apache.spark.sql.hive.HiveExternalCatalog" private def externalCatalogClassName(conf: SparkConf): String = { From d9e0ec73dc6a9685b65a08c8166129299f6c8960 Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 17:24:44 -0800 Subject: [PATCH 07/10] preserve the behavior --- .../scala/org/apache/spark/sql/internal/SharedState.scala | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index c82d6a781983f..4b989a6c213c8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -196,15 +196,14 @@ private[sql] class SharedState( object SharedState extends Logging { private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED) && factory.isEmpty) { - factory.synchronized { + synchronized { if (factory.isEmpty) { try { URL.setURLStreamHandlerFactory(defaultFactory) factory = Some(defaultFactory) } catch { - case NonFatal(e) => - logWarning("URL.setURLStreamHandlerFactory failed to set " + - "FsUrlStreamHandlerFactory", e) + case NonFatal(_) => + logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") } } } From c4ec2d5bde527462f3b955e4015178acde89ac1c Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 17:41:29 -0800 Subject: [PATCH 08/10] Use AtomicBoolean --- .../spark/sql/internal/SharedState.scala | 25 ++++++++----------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index 4b989a6c213c8..27d8b6f64b3e2 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.internal import java.net.URL import java.util.{Locale, UUID} import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean import javax.annotation.concurrent.GuardedBy import scala.reflect.ClassTag @@ -194,26 +195,20 @@ private[sql] class SharedState( } object SharedState extends Logging { + private val fsUrlStreamHandlerFactoryInitialized = new AtomicBoolean(false) + private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { - if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED) && factory.isEmpty) { - synchronized { - if (factory.isEmpty) { - try { - URL.setURLStreamHandlerFactory(defaultFactory) - factory = Some(defaultFactory) - } catch { - case NonFatal(_) => - logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") - } - } + if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED) && + fsUrlStreamHandlerFactoryInitialized.compareAndSet(false, true)) { + try { + URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) + } catch { + case NonFatal(_) => + logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") } } } - @volatile private var factory: Option[FsUrlStreamHandlerFactory] = None - - private lazy val defaultFactory = new FsUrlStreamHandlerFactory() - private val HIVE_EXTERNAL_CATALOG_CLASS_NAME = "org.apache.spark.sql.hive.HiveExternalCatalog" private def externalCatalogClassName(conf: SparkConf): String = { From cd9ed48d94dbbcd303396effe5666a96061a0264 Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 18:39:39 -0800 Subject: [PATCH 09/10] fix --- .../main/scala/org/apache/spark/sql/internal/SharedState.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index 27d8b6f64b3e2..f9f437df061f0 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -199,7 +199,7 @@ object SharedState extends Logging { private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED) && - fsUrlStreamHandlerFactoryInitialized.compareAndSet(false, true)) { + !fsUrlStreamHandlerFactoryInitialized.getAndSet(true)) { try { URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) } catch { From 755c9c0384d7cf016999fdd499800189e77ee733 Mon Sep 17 00:00:00 2001 From: Dongjoon Hyun Date: Sun, 17 Nov 2019 20:18:29 -0800 Subject: [PATCH 10/10] Revert AtomicBoolean --- .../spark/sql/internal/SharedState.scala | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala index f9f437df061f0..81a9c76511d8b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala @@ -20,7 +20,6 @@ package org.apache.spark.sql.internal import java.net.URL import java.util.{Locale, UUID} import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.atomic.AtomicBoolean import javax.annotation.concurrent.GuardedBy import scala.reflect.ClassTag @@ -195,16 +194,21 @@ private[sql] class SharedState( } object SharedState extends Logging { - private val fsUrlStreamHandlerFactoryInitialized = new AtomicBoolean(false) + @volatile private var fsUrlStreamHandlerFactoryInitialized = false private def setFsUrlStreamHandlerFactory(conf: SparkConf): Unit = { - if (conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED) && - !fsUrlStreamHandlerFactoryInitialized.getAndSet(true)) { - try { - URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) - } catch { - case NonFatal(_) => - logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") + if (!fsUrlStreamHandlerFactoryInitialized && + conf.get(DEFAULT_URL_STREAM_HANDLER_FACTORY_ENABLED)) { + synchronized { + if (!fsUrlStreamHandlerFactoryInitialized) { + try { + URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()) + fsUrlStreamHandlerFactoryInitialized = true + } catch { + case NonFatal(_) => + logWarning("URL.setURLStreamHandlerFactory failed to set FsUrlStreamHandlerFactory") + } + } } } }