Skip to content

[SPARK-26586][SS] Fix race condition that causes streams to run with unexpected confs #23513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mukulmurthy wants to merge 3 commits into from
+27 −2
Closed

[SPARK-26586][SS] Fix race condition that causes streams to run with unexpected confs #23513

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
10f0994
add failing test
mukulmurthy Jan 10, 2019
b35eef0
clone the spark session in StreamExecution constructor
mukulmurthy Jan 10, 2019
1319ad3
Ryan's PR comments
mukulmurthy Jan 10, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
clone the spark session in StreamExecution constructor
  • Loading branch information
@mukulmurthy
mukulmurthy committed Jan 10, 2019
commit b35eef0125964f29a5b029cbcf1609381e3c4f5a
5 changes: 3 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,9 @@ abstract class StreamExecution(
lazy val streamMetrics = new MetricsReporter(
this, s"spark.streaming.${Option(name).getOrElse(id)}")

/** Isolated spark session to run the batches with. */
private val sparkSessionForStream = sparkSession.cloneSession()

/**
* The thread that runs the micro-batches of this stream. Note that this thread must be
* [[org.apache.spark.util.UninterruptibleThread]] to workaround KAFKA-1894: interrupting a
Expand Down Expand Up @@ -270,8 +273,6 @@ abstract class StreamExecution(
// force initialization of the logical plan so that the sources can be created
logicalPlan

// Isolated spark session to run the batches with.
val sparkSessionForStream = sparkSession.cloneSession()
// Adaptive execution can change num shuffle partitions, disallow
sparkSessionForStream.conf.set(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, "false")
// Disable cost-based join optimization as we do not want stateful operations to be rearranged
Expand Down