[SC-5465] Throw errors for unsupported options in Kafka 0.8 source

zsxwing · zsxwing · commit d7c59b2d7cc0 · 2017-01-17T12:01:46.000-08:00
## What changes were proposed in this pull request? A follow up PR for apache#161 to disallow unsupported options. ## How was this patch tested? `test("unsupported options")` Author: Shixiong Zhu <shixiong@databricks.com> Closes apache#169 from zsxwing/kafka08-errors.
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/JsonUtils.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/JsonUtils.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaCluster.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaCluster.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSource.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSource.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSourceOffset.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSourceOffset.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSourceProvider.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSourceProvider.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
@@ -150,6 +150,13 @@ private[kafka08] class KafkaSourceProvider extends StreamSourceProvider
         s"Option 'kafka.bootstrap.servers' must be specified for " +
           s"configuring Kafka consumer")
     }
+
+    for (unsupportedOption <- UNSUPPORTED_OPTIONS) {
+      if (caseInsensitiveParams.contains(unsupportedOption.toLowerCase)) {
+        throw new IllegalArgumentException(
+          s"'$unsupportedOption' is not supported in Kafka 0.8 source")
+      }
+    }
   }
 
   override def shortName: String = "kafka08"
@@ -158,4 +165,5 @@ private[kafka08] class KafkaSourceProvider extends StreamSourceProvider
 private[kafka08] object KafkaSourceProvider {
   private val STRATEGY_OPTION_KEYS = Set("subscribe", "assign")
   private val STARTING_OFFSETS_OPTION_KEY = "startingoffsets"
+  private val UNSUPPORTED_OPTIONS = Set("subscribePattern", "failOnDataLoss")
 }
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSourceRDD.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/KafkaSourceRDD.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/StartingOffsets.scala b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/StartingOffsets.scala
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/package-info.java b/external/kafka-0-8-sql/src/main/scala/org/apache/spark/sql/kafka08/package-info.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 /*
- * Copyright © 2016 Databricks, Inc.
+ * Copyright (C) 2016 Databricks, Inc.
  *
  * Portions of this software incorporate or are derived from software contained within Apache Spark,
  * and this modified software differs from the Apache Spark software provided under the Apache
diff --git a/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/JsonUtilsSuite.scala b/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/JsonUtilsSuite.scala
@@ -14,6 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+/*
+ * Copyright (C) 2016 Databricks, Inc.
+ *
+ * Portions of this software incorporate or are derived from software contained within Apache Spark,
+ * and this modified software differs from the Apache Spark software provided under the Apache
+ * License, Version 2.0, a copy of which you may obtain at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
 
 package org.apache.spark.sql.kafka08
 
diff --git a/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/KafkaSourceOffsetSuite.scala b/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/KafkaSourceOffsetSuite.scala
@@ -14,6 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+/*
+ * Copyright (C) 2016 Databricks, Inc.
+ *
+ * Portions of this software incorporate or are derived from software contained within Apache Spark,
+ * and this modified software differs from the Apache Spark software provided under the Apache
+ * License, Version 2.0, a copy of which you may obtain at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
 
 package org.apache.spark.sql.kafka08
 
diff --git a/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/KafkaSourceSuite.scala b/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/KafkaSourceSuite.scala
@@ -14,6 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+/*
+ * Copyright (C) 2016 Databricks, Inc.
+ *
+ * Portions of this software incorporate or are derived from software contained within Apache Spark,
+ * and this modified software differs from the Apache Spark software provided under the Apache
+ * License, Version 2.0, a copy of which you may obtain at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
 
 package org.apache.spark.sql.kafka08
 
@@ -440,6 +448,26 @@ class KafkaSourceSuite extends KafkaSourceTest {
     query.stop()
   }
 
+  test("unsupported options") {
+    def testUnsupportedOption(key: String, value: String = "someValue"): Unit = {
+      val ex = intercept[IllegalArgumentException] {
+        val reader = spark
+          .readStream
+          .format("kafka08")
+          .option("subscribe", "topic")
+          .option("kafka.bootstrap.servers", "somehost")
+          .option(key, value)
+        reader.load()
+      }
+      for (msg <- Seq("0.8", "not supported", key)) {
+        assert(ex.getMessage.contains(msg))
+      }
+    }
+
+    testUnsupportedOption("subscribePattern")
+    testUnsupportedOption("failOnDataLoss")
+  }
+
   private def testFromLatestOffsets(
       topic: String,
       addPartitions: Boolean,
diff --git a/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/KafkaTestUtils.scala b/external/kafka-0-8-sql/src/test/scala/org/apache/spark/sql/kafka08/KafkaTestUtils.scala
@@ -14,6 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+/*
+ * Copyright (C) 2016 Databricks, Inc.
+ *
+ * Portions of this software incorporate or are derived from software contained within Apache Spark,
+ * and this modified software differs from the Apache Spark software provided under the Apache
+ * License, Version 2.0, a copy of which you may obtain at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
 
 package org.apache.spark.sql.kafka08
 

Original file line number	Diff line number	Diff line change
`@@ -15,7 +15,7 @@`
`15`	`15`	`* limitations under the License.`
`16`	`16`	`*/`
`17`	`17`	`/*`
`18`		`- * Copyright © 2016 Databricks, Inc.`
	`18`	`+ * Copyright (C) 2016 Databricks, Inc.`
`19`	`19`	`*`
`20`	`20`	`* Portions of this software incorporate or are derived from software contained within Apache Spark,`
`21`	`21`	`* and this modified software differs from the Apache Spark software provided under the Apache`