peter-toth commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3979183353
##########
connector/kafka-0-10-sql/src/main/resources/error/kafka-error-conditions.json:
##########
@@ -1,4 +1,9 @@
{
+ "KAFKA_DISALLOWED_OPTION" : {
+ "message" : [
+ "The Kafka option 'kafka.<option>' is not allowed because it is listed
in <config>."
Review Comment:
**Finding 3.** A user who hits this has nowhere to look it up.
`docs/streaming/structured-streaming-kafka-integration.md:1021` carries the
hand-written list of Kafka params that cannot be set, and this PR makes that
list operator-extendable. A sentence there would close it:
> In addition, an operator can reject further Kafka params by listing their
names, without the `kafka.` prefix, in `spark.sql.kafka.disallowedOptions`.
`.internal()` is not a reason to skip it.
`spark.sql.streaming.kafka.useDeprecatedOffsetFetching` is internal too and is
documented in the same guide at line 659.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3913,6 +3913,19 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val KAFKA_DISALLOWED_OPTIONS =
+ buildConf("spark.sql.kafka.disallowedOptions")
Review Comment:
**Finding 1.** `buildConf` makes this runtime-modifiable, so the application
this is meant to restrict can turn it off with `SET
spark.sql.kafka.disallowedOptions=`.
I measured it in a scratch suite against this head. With the conf set to
`sasl.jaas.config`, `kafkaParamsForProducer` rejects `kafka.sasl.jaas.config`.
After clearing the same conf, the same call passes the value straight through:
```scala
val params = CaseInsensitiveMap(Map(
"kafka.bootstrap.servers" -> "dummy", "kafka.sasl.jaas.config" -> "evil"))
val conf = new SQLConf()
conf.setConfString(SQLConf.KAFKA_DISALLOWED_OPTIONS.key, "sasl.jaas.config")
SQLConf.withExistingConf(conf) {
intercept[IllegalArgumentException](KafkaSourceProvider.kafkaParamsForProducer(params))
}
// what `SET spark.sql.kafka.disallowedOptions=` does to the session conf
conf.setConfString(SQLConf.KAFKA_DISALLOWED_OPTIONS.key, "")
SQLConf.withExistingConf(conf) {
assert(KafkaSourceProvider.kafkaParamsForProducer(params).get("sasl.jaas.config")
=== "evil")
}
```
Both assertions hold.
`conf.isModifiable("spark.sql.kafka.disallowedOptions")` is `true` for the same
reason, so `SET` reaches it.
`buildStaticConf` is the fix: static SQL configs are cross-session and an
external user can read but not set them. A connector-level
`spark.kafka.disallowedOptions` alongside `spark.kafka.consumer.cache.*` in
`connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/package.scala`
would also work.
If the session override is deliberate, then the `.doc` should say so,
because "operators control which options an application may set" reads as a
boundary today.
This is the same shape as the allowlist in #58613's sibling
https://github.com/apache/spark/pull/58614, so whichever way you go probably
applies to both.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3913,6 +3913,19 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val KAFKA_DISALLOWED_OPTIONS =
+ buildConf("spark.sql.kafka.disallowedOptions")
+ .internal()
+ .doc("A comma-separated list of Kafka client option names (without the
'kafka.' prefix) " +
+ "that are not allowed to be set through Kafka source/sink options.
Empty by default, " +
+ "which allows all options and preserves the previous behavior; when
non-empty, setting a " +
+ "listed option raises an error.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .stringConf
+ .toSequence
+ .createWithDefault(Nil)
Review Comment:
**Finding 2.** An operator who writes the name the way it appears everywhere
else gets no error and no protection:
```
spark.sql.kafka.disallowedOptions=kafka.sasl.jaas.config
```
The entries are matched against the *stripped* name, so a `kafka.`-prefixed
entry never matches and the option goes through. I measured it: with that conf
value, `kafkaParamsForProducer` returns `sasl.jaas.config` unchanged. The error
message prints `'kafka.<option>'`, so copying the name out of the error is the
natural way to land here.
A denylist should not fail open on that. Rejecting the prefixed form at
parse time is two lines. I applied and ran this, and `setConfString(key,
"kafka.max.poll.records")` then throws:
```suggestion
.stringConf
.toSequence
.checkValue(_.forall(!_.toLowerCase(Locale.ROOT).startsWith("kafka.")),
"Kafka option names must be listed without the 'kafka.' prefix.")
.createWithDefault(Nil)
```
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceProviderSuite.scala:
##########
@@ -154,4 +156,45 @@ 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"))
+ SQLConf.withExistingConf(conf) {
+ val e = intercept[IllegalArgumentException] {
+ getKafkaDataSourceScan(options).toBatch()
+ }
+ assert(e.getMessage.contains("kafka.max.poll.records"))
+ }
+ }
+
+ test("SPARK-59328: disallowed Kafka options are rejected on the sink path") {
+ val sparkEnv = mock(classOf[SparkEnv])
+ when(sparkEnv.conf).thenReturn(new SparkConf())
+ SparkEnv.set(sparkEnv)
+
Review Comment:
**Finding 5.** `kafkaParamsForProducer` never reads `SparkEnv`.
`KafkaConfigUpdater.build()` just returns the map, and `redactParams` runs only
under `isDebugEnabled` and tolerates a null `SparkEnv` anyway. I dropped these
lines and the test still passes.
```suggestion
```
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceProviderSuite.scala:
##########
@@ -154,4 +156,45 @@ 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"))
+ SQLConf.withExistingConf(conf) {
+ val e = intercept[IllegalArgumentException] {
+ getKafkaDataSourceScan(options).toBatch()
+ }
+ assert(e.getMessage.contains("kafka.max.poll.records"))
Review Comment:
**Finding 4.** `getMessage.contains` leaves the new condition name and both
message parameters unpinned. `KafkaIllegalArgumentException` mixes in
`SparkThrowable`, so `checkError` works here. I ran this against this head:
```scala
checkError(
exception = intercept[KafkaIllegalArgumentException] {
getKafkaDataSourceScan(options).toBatch()
},
condition = "KAFKA_DISALLOWED_OPTION",
parameters = Map(
"option" -> "max.poll.records",
"config" -> "spark.sql.kafka.disallowedOptions"))
```
Same for the sink test at line 197.
--
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]
