You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "vrozov (via GitHub)" <gi...@apache.org> on 2025/03/21 17:52:02 UTC
[PR] [SPARK-51586][SS] initialize input partitions independent of columnar support [spark]
vrozov opened a new pull request, #50348:
URL: https://github.com/apache/spark/pull/50348
### What changes were proposed in this pull request?
Initialize input partitions independent of columnar support in `ContinuousScanExec`
### Why are the changes needed?
After [SPARK-45080](https://issues.apache.org/jira/browse/SPARK-45080) and [PR 42823](https://github.com/apache/spark/pull/42823) Kafka continuous stream may go into infinite loop of reconfiguring. As `KafkaScan` columnar support mode is hardcoded to `UNSUPPORTED` `ContinuousScanExec` does not initialize `inputPartitions` in `supportsColumnar` as it did prior to the above PR. So, there is no call to `KafkaContinuousStream.planInputPartitions` during query planning. It leaves `KafkaContinuousStream.knownPartitions` uninitialized, so call to `needsReconfiguration` returns `true`. Now `epochUpdateThread` in `ContinuousExecution` requests interruption of `queryExecutionThread`, so that thread also can't initialize `knownPartitions` as it checks for interrupts in `KafkaOffsetReaderAdmin.withRetries` and exits before `knownPartitions` is assigned.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes bug that may prevent kafka continuous stream from working properly
### How was this patch tested?
Using existing Kafka tests. Those tests should complete faster.
### Was this patch authored or co-authored using generative AI tooling?
No
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "vrozov (via GitHub)" <gi...@apache.org>.
vrozov commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2010314980
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
IMO, it is too small to have a different PR and it is somewhat related to the Kafka issue. Let me know if you strongly prefer to revert this change.
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2010917948
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
I prefer to revert the change. I don't want to mess up the commit log.
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2009331917
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
nit: shall we not change the code which is not relevant? You can have a different PR to do that, but please find more cases and change altogether.
--
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-51586][SS] initialize input partitions independent of columnar support in continuous mode [spark]
Posted by "vrozov (via GitHub)" <gi...@apache.org>.
vrozov commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2011179103
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
https://docs.scala-lang.org/scala3/reference/dropped-features/auto-apply.html: `Thread.currentThread` still will be the recommended way for Scala 3 as well.
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2011147337
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
I think this is arguably not super clear style rule. Only "pure" parameterless method can remove the paren from the best practice of Scala (it's not that you can remove the paren as long as the method does not take any param), but there are arguments of what is "pure" method.
Here is the post which starts with very simple question got distracted with what is pure method.
https://users.scala-lang.org/t/paramterless-functions-with-and-without-parentheses-different/9939/3
In addition, Scala 3 will enforce the defined method with paren to "always" call with paren, regardless of parameterless, "pure" method, etcetc.
So I don't think this is an obvious style fix - someone could argue that we are going to break the style.
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "vrozov (via GitHub)" <gi...@apache.org>.
vrozov commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2011034667
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
OK, reverted. IMO, it will be good to fix the style anyway and enforce it.
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2011147337
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
I think this is arguably not super clear style rule. Only "pure" parameterless method can remove the paren from the best practice of Scala (it's not that you can remove the paren as long as the method does not take any param), but there are arguments of what is "pure" method.
Here is the post which starts with very simple question got distracted with what is pure method.
https://users.scala-lang.org/t/paramterless-functions-with-and-without-parentheses-different/9939/3
In addition, Scala 3 will enforce the defined method with paren to "always" call with paren, regardless of parameterless, "pure" method, etcetc.
So I don't think this is definitely a style fix - someone could argue that we are going to break the style.
--
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-51586][SS] initialize input partitions independent of columnar support in continuous mode [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2011212796
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
> The latter is idiomatic Scala because it conforms to the uniform access principle. This principle states that one should be able to change an object member from a field to a non-side-effecting method and back without affecting clients that access the member. Consequently, Scala encourages to define such "property" methods without a () parameter list whereas side-effecting methods should be defined with it. Methods defined in Java cannot make this distinction; for them a () is always mandatory. So Scala fixes the problem on the client side, by allowing the parameterless references. But where Scala allows that freedom for all method references, Scala 3 restricts it to references of external methods that are not defined themselves in Scala 3.
If you see the origin thread, the definition of "pure" method is not super obvious; it's not something compiler would be able to catch and it's mostly on human's justification. This is just a homework Scala left to us and I don't really like the fact Scala makes such an ambiguity.
Continuing from there, I'd rather interpret this as Java is just excluded from the enforcement. It is still arguable what is "pure" method and everyone has every different opinions. Thread.currentThread is a native method, and it's not super obvious that there is no side-effect.
--
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-51586][SS] initialize input partitions independent of columnar support in continuous mode [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50348:
URL: https://github.com/apache/spark/pull/50348#issuecomment-2749812507
Thanks! Merging to master.
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "vrozov (via GitHub)" <gi...@apache.org>.
vrozov commented on PR #50348:
URL: https://github.com/apache/spark/pull/50348#issuecomment-2744063797
@HeartSaVioR please review
--
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-51586][SS] initialize input partitions independent of columnar support [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50348:
URL: https://github.com/apache/spark/pull/50348#discussion_r2010917948
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -535,7 +535,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
var attempt = 1
var lastException: Throwable = null
while (result.isEmpty && attempt <= maxOffsetFetchAttempts
- && !Thread.currentThread().isInterrupted) {
+ && !Thread.currentThread.isInterrupted) {
Review Comment:
I prefer to revert the change. I don't want to mess up the commit log. What you change is completely just a style issue and I don't want to give this line to be shown as "fixing bug".
--
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-51586][SS] initialize input partitions independent of columnar support in continuous mode [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #50348: [SPARK-51586][SS] initialize input partitions independent of columnar support in continuous mode
URL: https://github.com/apache/spark/pull/50348
--
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