2 files changed, 18 insertions, 1 deletions
diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala
index 16423e771a..f9c01f30f1 100644
--- a/core/src/main/scala/org/apache/spark/SparkConf.scala
+++ b/core/src/main/scala/org/apache/spark/SparkConf.scala
@@ -388,7 +388,10 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
   def getAppId: String = get("spark.app.id")
 
   /** Does the configuration contain a given parameter? */
-  def contains(key: String): Boolean = settings.containsKey(key)
+  def contains(key: String): Boolean = {
+    settings.containsKey(key) ||
+      configsWithAlternatives.get(key).toSeq.flatten.exists { alt => contains(alt.key) }
+  }
 
   /** Copy this object */
   override def clone: SparkConf = {
diff --git a/core/src/test/scala/org/apache/spark/SparkConfSuite.scala b/core/src/test/scala/org/apache/spark/SparkConfSuite.scala
index 79881f30b2..a883d1b57e 100644
--- a/core/src/test/scala/org/apache/spark/SparkConfSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkConfSuite.scala
@@ -267,6 +267,20 @@ class SparkConfSuite extends SparkFunSuite with LocalSparkContext with ResetSyst
     conf.set("spark.akka.lookupTimeout", "4")
     assert(RpcUtils.lookupRpcTimeout(conf).duration === (4 seconds))
   }
+
+  test("SPARK-13727") {
+    val conf = new SparkConf()
+    // set the conf in the deprecated way
+    conf.set("spark.io.compression.lz4.block.size", "12345")
+    // get the conf in the recommended way
+    assert(conf.get("spark.io.compression.lz4.blockSize") === "12345")
+    // we can still get the conf in the deprecated way
+    assert(conf.get("spark.io.compression.lz4.block.size") === "12345")
+    // the contains() also works as expected
+    assert(conf.contains("spark.io.compression.lz4.block.size"))
+    assert(conf.contains("spark.io.compression.lz4.blockSize"))
+    assert(conf.contains("spark.io.unknown") === false)
+  }
 }
 
 class Class1 {}