peter-toth commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3987141550
##########
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:
**Finding 6.** The follow-up added two guards and the suite pins neither.
`setConf(entry, value)` stores `entry.stringConverter(value)`
(`SQLConf.scala:10002`), and `checkValue` only wraps the String -> T direction:
`transform` rebuilds `converter` and leaves `stringConverter` alone
(`ConfigBuilder.scala:162`). So this line skips the new prefix check. The tests
also never reach a reader of `isStaticConfigKey`, so `buildStaticConf` is
equally invisible to them. Swap it back to `buildConf` and both new cases still
pass.
Both guards fit in the existing suite with no session:
```scala
test("SPARK-59328: the disallowed-options denylist is an operator
boundary") {
// Static, so an application cannot SET it away. This was `true` before
the follow-up.
assert(!new SQLConf().isModifiable(SQLConf.KAFKA_DISALLOWED_OPTIONS.key))
// Entries carrying the "kafka." prefix are rejected, so the denylist
cannot fail open.
checkError(
exception = intercept[SparkIllegalArgumentException] {
new SQLConf().setConfString(
SQLConf.KAFKA_DISALLOWED_OPTIONS.key, "kafka.max.poll.records")
},
condition = "INVALID_CONF_VALUE.REQUIREMENT",
parameters = Map(
"confName" -> "spark.sql.kafka.disallowedOptions",
"confValue" -> "kafka.max.poll.records",
"confRequirement" -> "Kafka option names must be listed without the
'kafka.' prefix."))
}
```
`setConfString` is the path that runs `valueConverter`
(`SQLConf.scala:9992`), and `configRequirementError` builds a
`SparkIllegalArgumentException` with `INVALID_CONF_VALUE.REQUIREMENT`
(`ConfigBuilder.scala:132`), so `checkError` reads it the same way your two new
cases read `KAFKA_DISALLOWED_OPTION`. I derived this from reading those call
sites rather than running 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:
**Finding 7.** The paragraph doesn't say where the config has to be set. It
is a `buildStaticConf` now, so `SET spark.sql.kafka.disallowedOptions=...` and
`spark.conf.set(...)` both fail with `CANNOT_MODIFY_STATIC_CONFIG`, and that is
the first thing an operator reading this will try. Being static is also what
makes this a boundary rather than a hint, so it belongs in the text.
```suggestion
param through a Kafka source or sink option will then throw an exception.
This is a static
configuration. It must be set when the SparkSession is created, for example
with
`--conf spark.sql.kafka.disallowedOptions=...`, and an application cannot
change it 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:
**Finding 8.** This is the only `buildStaticConf` call in `SQLConf.scala`.
The other 38 static SQL configs live in `StaticSQLConf`, which imports the
builder from here for that purpose (`StaticSQLConf.scala:37`). Nothing breaks
where it is, since `onCreate` registers the static key either way.
Someone auditing which SQL configs are static reads `StaticSQLConf` though,
and this one is invisible there. Moving it is a cut and paste plus a `SQLConf.`
-> `StaticSQLConf.` rename at the three call sites
(`KafkaSourceProvider.scala:813`, `:821`, 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]
