HyukjinKwon commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3994455941
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceProviderSuite.scala:
##########
@@ -154,4 +156,49 @@ class KafkaSourceProviderSuite extends SparkFunSuite {
}
}
}
+
+ test("SPARK-59328: disallowed Kafka options are rejected on the source
path") {
+ // KafkaBatch reads its default poll timeout from SparkEnv, so provide a
mock one.
+ val sparkEnv = mock(classOf[SparkEnv])
+ when(sparkEnv.conf).thenReturn(new SparkConf())
+ SparkEnv.set(sparkEnv)
+
+ val options =
buildKafkaSourceCaseInsensitiveStringMap("kafka.max.poll.records" -> "1")
+ // Empty denylist (the default) preserves the previous behavior: the
option is accepted.
+ getKafkaDataSourceScan(options).toBatch()
+ // When the option name is denylisted, building the batch scan is rejected.
+ val conf = new SQLConf()
+ conf.setConf(SQLConf.KAFKA_DISALLOWED_OPTIONS, Seq("max.poll.records"))
Review Comment:
Done in 99308fc. Added `SPARK-59328: the disallowed-options denylist is an
operator boundary`, which asserts `!new SQLConf().isModifiable(...)` and that a
`kafka.`-prefixed entry via `setConfString` throws
`INVALID_CONF_VALUE.REQUIREMENT`. Reverting `buildStaticConf` or the prefix
`checkValue` now fails it.
##########
docs/streaming/structured-streaming-kafka-integration.md:
##########
@@ -1047,6 +1047,10 @@ DataFrame operations to explicitly serialize the values
into either strings or b
- **interceptor.classes**: Kafka source always read keys and values as byte
arrays. It's not safe to
use ConsumerInterceptor as it may break the query.
+In addition, an operator can reject further Kafka params by listing their
names, without the
+`kafka.` prefix, in the `spark.sql.kafka.disallowedOptions` configuration;
setting any listed
+param through a Kafka source or sink option will then throw an exception.
Review Comment:
Applied your suggestion in 99308fc -- the guide now says the config is
static and must be set when the SparkSession is created (`--conf
spark.sql.kafka.disallowedOptions=...`), not at runtime.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3913,6 +3913,23 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val KAFKA_DISALLOWED_OPTIONS =
+ buildStaticConf("spark.sql.kafka.disallowedOptions")
Review Comment:
Moved to `StaticSQLConf` in 99308fc, with the `SQLConf.` -> `StaticSQLConf.`
rename at the three call sites (`KafkaSourceProvider` x2 and the suite).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
