[spark] branch master updated: [SPARK-28907][CORE] Review invalid usage of new Configuration()

[sr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

srowen 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 ca71177  [SPARK-28907][CORE] Review invalid usage of new Configuration()
ca71177 is described below

commit ca711778683a16999560cbdd7c61d98ad6bde6db
Author: Xianjin YE <ad...@gmail.com>
AuthorDate: Wed Sep 4 19:52:19 2019 -0500

    [SPARK-28907][CORE] Review invalid usage of new Configuration()
    
    ### What changes were proposed in this pull request?
    Replaces some incorrect usage of `new Configuration()` as it will load default configs defined in Hadoop
    
    ### Why are the changes needed?
    Unexpected config could be accessed instead of the expected config, see SPARK-28203 for example
    
    ### Does this PR introduce any user-facing change?
    No.
    
    ### How was this patch tested?
    Existed tests.
    
    Closes #25616 from advancedxy/remove_invalid_configuration.
    
    Authored-by: Xianjin YE <ad...@gmail.com>
    Signed-off-by: Sean Owen <se...@databricks.com>
---
 .../scala/org/apache/spark/input/PortableDataStream.scala   |  2 +-
 .../org/apache/spark/input/WholeTextFileRecordReader.scala  | 13 +++++++++++--
 .../execution/datasources/HadoopFileWholeTextReader.scala   |  1 +
 3 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/input/PortableDataStream.scala b/core/src/main/scala/org/apache/spark/input/PortableDataStream.scala
index 5b33c11..6a4af01 100644
--- a/core/src/main/scala/org/apache/spark/input/PortableDataStream.scala
+++ b/core/src/main/scala/org/apache/spark/input/PortableDataStream.scala
@@ -172,7 +172,7 @@ class PortableDataStream(
 
   @transient private lazy val conf = {
     val bais = new ByteArrayInputStream(confBytes)
-    val nconf = new Configuration()
+    val nconf = new Configuration(false)
     nconf.readFields(new DataInputStream(bais))
     nconf
   }
diff --git a/core/src/main/scala/org/apache/spark/input/WholeTextFileRecordReader.scala b/core/src/main/scala/org/apache/spark/input/WholeTextFileRecordReader.scala
index 6b7f0866..28fd1ff1b 100644
--- a/core/src/main/scala/org/apache/spark/input/WholeTextFileRecordReader.scala
+++ b/core/src/main/scala/org/apache/spark/input/WholeTextFileRecordReader.scala
@@ -69,7 +69,7 @@ private[spark] class WholeTextFileRecordReader(
 
   override def nextKeyValue(): Boolean = {
     if (!processed) {
-      val conf = new Configuration
+      val conf = getConf
       val factory = new CompressionCodecFactory(conf)
       val codec = factory.getCodec(path)  // infers from file ext.
       val fileIn = fs.open(path)
@@ -108,8 +108,17 @@ private[spark] class ConfigurableCombineFileRecordReader[K, V](
   override def initNextRecordReader(): Boolean = {
     val r = super.initNextRecordReader()
     if (r) {
-      this.curReader.asInstanceOf[HConfigurable].setConf(getConf)
+      if (getConf != null) {
+        this.curReader.asInstanceOf[HConfigurable].setConf(getConf)
+      }
     }
     r
   }
+
+  override def setConf(c: Configuration): Unit = {
+    super.setConf(c)
+    if (this.curReader != null) {
+      this.curReader.asInstanceOf[HConfigurable].setConf(c)
+    }
+  }
 }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/HadoopFileWholeTextReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/HadoopFileWholeTextReader.scala
index f5724f7..0e6d803 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/HadoopFileWholeTextReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/HadoopFileWholeTextReader.scala
@@ -45,6 +45,7 @@ class HadoopFileWholeTextReader(file: PartitionedFile, conf: Configuration)
     val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
     val hadoopAttemptContext = new TaskAttemptContextImpl(conf, attemptId)
     val reader = new WholeTextFileRecordReader(fileSplit, hadoopAttemptContext, 0)
+    reader.setConf(hadoopAttemptContext.getConfiguration)
     reader.initialize(fileSplit, hadoopAttemptContext)
     new RecordReaderIterator(reader)
   }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org