Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/15102#discussion_r80565318
--- Diff:
external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
---
@@ -0,0 +1,263 @@
+/*
+ * 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.kafka010
+
+import java.{util => ju}
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.common.serialization.ByteArrayDeserializer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.execution.streaming.Source
+import org.apache.spark.sql.kafka010.KafkaSource._
+import org.apache.spark.sql.sources.{DataSourceRegister,
StreamSourceProvider}
+import org.apache.spark.sql.types.StructType
+
+/**
+ * The provider class for the [[KafkaSource]]. This provider is designed
such that it throws
+ * IllegalArgumentException when the Kafka Dataset is created, so that it
can catch
+ * missing options even before the query is started.
+ */
+private[kafka010] class KafkaSourceProvider extends StreamSourceProvider
+ with DataSourceRegister with Logging {
+ import KafkaSourceProvider._
+
+ /**
+ * Returns the name and schema of the source. In addition, it also
verifies whether the options
+ * are correct and sufficient to create the [[KafkaSource]] when the
query is started.
+ */
+ override def sourceSchema(
+ sqlContext: SQLContext,
+ schema: Option[StructType],
+ providerName: String,
+ parameters: Map[String, String]): (String, StructType) = {
+ validateOptions(parameters)
+ ("kafka", KafkaSource.kafkaSchema)
+ }
+
+ override def createSource(
+ sqlContext: SQLContext,
+ metadataPath: String,
+ schema: Option[StructType],
+ providerName: String,
+ parameters: Map[String, String]): Source = {
+ validateOptions(parameters)
+ val caseInsensitiveParams = parameters.map { case (k, v) =>
(k.toLowerCase, v) }
+ val specifiedKafkaParams =
+ parameters
+ .keySet
+ .filter(_.toLowerCase.startsWith("kafka."))
+ .map { k => k.drop(6).toString -> parameters(k) }
+ .toMap
+
+ val deserClassName = classOf[ByteArrayDeserializer].getName
+ val uniqueGroupId =
s"spark-kafka-source-${UUID.randomUUID}-${metadataPath.hashCode}"
+
+ val autoOffsetResetValue =
caseInsensitiveParams.get(STARTING_OFFSET_OPTION_KEY) match {
+ case Some(value) => value.trim() // same values as those supported
by auto.offset.reset
+ case None => "latest"
+ }
+
+ val kafkaParamsForStrategy =
+ ConfigUpdater("source", specifiedKafkaParams)
+ .set(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, deserClassName)
+ .set(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
deserClassName)
+
+ // So that consumers in Kafka source do not mess with any existing
group id
+ .set(ConsumerConfig.GROUP_ID_CONFIG, s"$uniqueGroupId-driver")
+
+ // So that consumers can start from earliest or latest
+ .set(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetResetValue)
+
+ // So that consumers in the driver does not commit offsets
unnecessaribly
+ .set(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+
+ // So that the driver does not pull too much data
+ .set(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, new
java.lang.Integer(1))
+
+ // If buffer config is not set, set it to reasonable value to work
around
+ // buffer issues (see KAFKA-3135)
+ .setIfUnset(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 65536:
java.lang.Integer)
+ .build()
+
+ val kafkaParamsForExecutors =
+ ConfigUpdater("executor", specifiedKafkaParams)
+ .set(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, deserClassName)
+ .set(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
deserClassName)
+
+ // So that consumers in executors never throw
NoOffsetForPartitionException
+ .set(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none")
+
+ // So that consumers in executors do not mess with any existing
group id
+ .set(ConsumerConfig.GROUP_ID_CONFIG, s"$uniqueGroupId-executor")
+
+ // So that consumers in executors does not commit offsets
unnecessaribly
+ .set(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+
+ // If buffer config is not set, set it to reasonable value to work
around
+ // buffer issues (see KAFKA-3135)
+ .setIfUnset(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 65536:
java.lang.Integer)
+ .build()
+
+ val strategy = caseInsensitiveParams.find(x =>
STRATEGY_OPTION_KEYS.contains(x._1)).get match {
+ case ("subscribe", value) =>
+ SubscribeStrategy(
+ value.split(",").map(_.trim()).filter(_.nonEmpty),
+ kafkaParamsForStrategy)
+ case ("subscribepattern", value) =>
+ SubscribePatternStrategy(
+ value.trim(),
+ kafkaParamsForStrategy)
+ case _ =>
+ // Should never reach here as we are already matching on
+ // matched strategy names
+ throw new IllegalArgumentException("Unknown option")
+ }
+
+ new KafkaSource(sqlContext, strategy, kafkaParamsForExecutors,
parameters)
+ }
+
+ private def validateOptions(parameters: Map[String, String]): Unit = {
+
+ // Validate source options
+
+ val caseInsensitiveParams = parameters.map { case (k, v) =>
(k.toLowerCase, v) }
+ val specifiedStrategies =
+ caseInsensitiveParams.filter { case (k, _) =>
STRATEGY_OPTION_KEYS.contains(k) }.toSeq
+ if (specifiedStrategies.isEmpty) {
+ throw new IllegalArgumentException(
+ "One of the following options must be specified for Kafka source: "
+ + STRATEGY_OPTION_KEYS.mkString(", ") + ". See the docs for more
details.")
+ } else if (specifiedStrategies.size > 1) {
+ throw new IllegalArgumentException(
+ "Only one of the following options can be specified for Kafka
source: "
+ + STRATEGY_OPTION_KEYS.mkString(", ") + ". See the docs for more
details.")
+ }
+
+ val strategy = caseInsensitiveParams.find(x =>
STRATEGY_OPTION_KEYS.contains(x._1)).get match {
+ case ("subscribe", value) =>
+ val topics = value.split(",").map(_.trim).filter(_.nonEmpty)
+ if (topics.isEmpty) {
+ throw new IllegalArgumentException(
+ "No topics to subscribe to as specified value for option " +
+ s"'subscribe' is '$value'")
+ }
+ case ("subscribepattern", value) =>
+ val pattern = caseInsensitiveParams("subscribepattern").trim()
+ if (pattern.isEmpty) {
+ throw new IllegalArgumentException(
+ "Pattern to subscribe is empty as specified value for option "
+
+ s"'subscribePattern' is '$value'")
+ }
+ case _ =>
+ // Should never reach here as we are already matching on
+ // matched strategy names
+ throw new IllegalArgumentException("Unknown option")
+ }
+
+ caseInsensitiveParams.get(STARTING_OFFSET_OPTION_KEY) match {
+ case Some(pos) if
!STARTING_OFFSET_OPTION_VALUES.contains(pos.trim.toLowerCase) =>
+ throw new IllegalArgumentException(
+ s"Illegal value '$pos' for option '$STARTING_OFFSET_OPTION_KEY',
" +
+ s"acceptable values are:
${STARTING_OFFSET_OPTION_VALUES.mkString(", ")}")
+ case _ =>
+ }
+
+ // Validate user-specified Kafka options
+
+ if
(caseInsensitiveParams.contains(s"kafka.${ConsumerConfig.GROUP_ID_CONFIG}")) {
+ throw new IllegalArgumentException(
+ s"Kafka option '${ConsumerConfig.GROUP_ID_CONFIG}' is not
supported as " +
+ s"user-specified consumer groups is not used to track offsets.")
+ }
+
+ if
(caseInsensitiveParams.contains(s"kafka.${ConsumerConfig.AUTO_OFFSET_RESET_CONFIG}"))
{
+ throw new IllegalArgumentException(
+ s"""
+ |Kafka option '${ConsumerConfig.AUTO_OFFSET_RESET_CONFIG}' is
not supported.
--- End diff --
As worded this sounds like a limitation that we plan to fix. It might be
better to say something along the lines of "Structured Streaming manages which
offsets are consumed internally, rather than rely on the kafka Consumer to do
it." I'd also clarify that `STARTING_OFFSET_OPTION_KEY` only applies when a
new Streaming query is started, and that resuming will always pick up from
where the query left off.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
