You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HyukjinKwon (via GitHub)" <gi...@apache.org> on 2026/09/08 12:44:29 UTC
[PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
HyukjinKwon opened a new pull request, #58613:
URL: https://github.com/apache/spark/pull/58613
### What changes were proposed in this pull request?
Add an opt-in, internal `spark.sql.kafka.disallowedOptions` (default empty). When set to a
comma-separated list of Kafka client option names (without the `kafka.` prefix), those options
are rejected if passed as `kafka.<name>` on the Kafka source or sink; an empty list (the default)
allows all options and preserves the previous behavior. The rejection is raised through the Kafka
connector's error-class framework (a new `KAFKA_DISALLOWED_OPTION` condition).
This is one of three PRs split out of a combined data-source-options change; the Avro schema-URL
and Hive class-loading parts are handled separately.
### Why are the changes needed?
Gives operators an optional control over which Kafka client options an application may set through
the data source. It defaults to empty, so there is no behavior change unless configured.
### Does this PR introduce _any_ user-facing change?
No by default. When `spark.sql.kafka.disallowedOptions` is set, a listed `kafka.<name>` option on a
Kafka read or write is rejected with a clear error.
### How was this patch tested?
New `KafkaSourceProviderSuite` cases: a listed option is rejected on both the source and the sink
path, and the empty default (unset) still accepts the option.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
This pull request and its description were written by Isaac.
Co-authored-by: Isaac <no...@databricks.com>
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3959701291
##########
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")
Review Comment:
4.4.0?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3966403527
##########
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")
Review Comment:
Sounds good just wanted to confirm, tnx!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3962605432
##########
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")
Review Comment:
Yes, this is intentionally targeting 4.3.0 rather than 4.4.0. The rationale is being discussed on a private channel, so I can't share the details here -- happy to follow up offline. Thanks for checking.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #58613:
URL: https://github.com/apache/spark/pull/58613#discussion_r3962605432
##########
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")
Review Comment:
Yes, this is intentionally targeting 4.3.0 rather than 4.4.0. The rationale is being discussed on a private channel, so I can't share the details here per ASF policy -- happy to follow up offline. Thanks for checking.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #58613:
URL: https://github.com/apache/spark/pull/58613#issuecomment-5656699796
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/973bed5ecc49b167cba2f7c4c8c146f05565a6c6
- merged into branch-4.x https://github.com/apache/spark/commit/61d94d56b26ce90d2c148b54a12f8042f7bb4559
- merged into branch-4.3 https://github.com/apache/spark/commit/9f652833af28fda7924649c402c1ff51fecb154c
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #58613: [SPARK-59328][SQL] Add an option to reject specified Kafka options in the Kafka data source
URL: https://github.com/apache/spark/pull/58613
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org