Github user koeninger commented on a diff in the pull request:
https://github.com/apache/spark/pull/15102#discussion_r81622511
--- Diff:
external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
---
@@ -0,0 +1,281 @@
+/*
+ * 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) = {
+ require(schema.isEmpty, "Kafka source has a fixed schema and cannot be
set with a custom one")
+ 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
+ // Each running query should use its own group id. Otherwise, the
query may be only assigned
+ // partial data since Kafka will assign partitions to multiple
consumers having the same group
+ // id. Hence, we should generate a unique id for each query.
+ 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
unnecessarily
+ .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)
+
+ // Make sure executors do only what the driver tells them.
+ .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
unnecessarily
+ .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")
+ }
+
+ val failOnCorruptMetadata =
+ caseInsensitiveParams.getOrElse(FAIL_ON_CORRUPT_METADATA_OPTION_KEY,
"false").toBoolean
--- End diff --
This is still defaulting to false, but the docs say (and I believe it
should) default to true.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
