This is an automated email from the ASF dual-hosted git repository.
vanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new c4bbfd1 [SPARK-24063][SS] Add maximum epoch queue threshold for
ContinuousExecution
c4bbfd1 is described below
commit c4bbfd177b4e7cb46f47b39df9fd71d2d9a12c6d
Author: Gabor Somogyi <gabor.g.somo...@gmail.com>
AuthorDate: Wed Feb 27 09:52:43 2019 -0800
[SPARK-24063][SS] Add maximum epoch queue threshold for ContinuousExecution
## What changes were proposed in this pull request?
Continuous processing is waiting on epochs which are not yet complete (for
example one partition is not making progress) and stores pending items in
queues. These queues are unbounded and can consume up all the memory easily. In
this PR I've added `spark.sql.streaming.continuous.epochBacklogQueueSize`
configuration possibility to make them bounded. If the related threshold
reached then the query will stop with `IllegalStateException`.
## How was this patch tested?
Existing + additional unit tests.
Closes #23156 from gaborgsomogyi/SPARK-24063.
Authored-by: Gabor Somogyi <gabor.g.somo...@gmail.com>
Signed-off-by: Marcelo Vanzin <van...@cloudera.com>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 10 +++
.../streaming/continuous/ContinuousExecution.scala | 38 ++++++++++
.../streaming/continuous/EpochCoordinator.scala | 20 ++++++
.../sql/streaming/continuous/ContinuousSuite.scala | 31 +++++++++
.../continuous/EpochCoordinatorSuite.scala | 81 +++++++++++++++++++++-
5 files changed, 177 insertions(+), 3 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 2cf471e..4d31bcd 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1436,6 +1436,13 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE =
+ buildConf("spark.sql.streaming.continuous.epochBacklogQueueSize")
+ .doc("The max number of entries to be stored in queue to wait for late
epochs. " +
+ "If this parameter is exceeded by the size of the queue, stream will
stop with an error.")
+ .intConf
+ .createWithDefault(10000)
+
val CONTINUOUS_STREAMING_EXECUTOR_QUEUE_SIZE =
buildConf("spark.sql.streaming.continuous.executorQueueSize")
.internal()
@@ -2066,6 +2073,9 @@ class SQLConf extends Serializable with Logging {
def literalPickMinimumPrecision: Boolean =
getConf(LITERAL_PICK_MINIMUM_PRECISION)
+ def continuousStreamingEpochBacklogQueueSize: Int =
+ getConf(CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE)
+
def continuousStreamingExecutorQueueSize: Int =
getConf(CONTINUOUS_STREAMING_EXECUTOR_QUEUE_SIZE)
def continuousStreamingExecutorPollIntervalMs: Long =
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
index 20101c7..3f88bcf 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.streaming.continuous
import java.util.UUID
import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
import java.util.function.UnaryOperator
import scala.collection.JavaConverters._
@@ -58,6 +59,9 @@ class ContinuousExecution(
// For use only in test harnesses.
private[sql] var currentEpochCoordinatorId: String = _
+ // Throwable that caused the execution to fail
+ private val failure: AtomicReference[Throwable] = new
AtomicReference[Throwable](null)
+
override val logicalPlan: LogicalPlan = {
val v2ToRelationMap = MutableMap[StreamingRelationV2,
StreamingDataSourceV2Relation]()
var nextSourceId = 0
@@ -261,6 +265,11 @@ class ContinuousExecution(
lastExecution.toRdd
}
}
+
+ val f = failure.get()
+ if (f != null) {
+ throw f
+ }
} catch {
case t: Throwable if StreamExecution.isInterruptionException(t,
sparkSession.sparkContext) &&
state.get() == RECONFIGURING =>
@@ -374,6 +383,35 @@ class ContinuousExecution(
}
/**
+ * Stores error and stops the query execution thread to terminate the query
in new thread.
+ */
+ def stopInNewThread(error: Throwable): Unit = {
+ if (failure.compareAndSet(null, error)) {
+ logError(s"Query $prettyIdString received exception $error")
+ stopInNewThread()
+ }
+ }
+
+ /**
+ * Stops the query execution thread to terminate the query in new thread.
+ */
+ private def stopInNewThread(): Unit = {
+ new Thread("stop-continuous-execution") {
+ setDaemon(true)
+
+ override def run(): Unit = {
+ try {
+ ContinuousExecution.this.stop()
+ } catch {
+ case e: Throwable =>
+ logError(e.getMessage, e)
+ throw e
+ }
+ }
+ }.start()
+ }
+
+ /**
* Stops the query execution thread to terminate the query.
*/
override def stop(): Unit = {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala
index a998422..decf524 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala
@@ -123,6 +123,9 @@ private[continuous] class EpochCoordinator(
override val rpcEnv: RpcEnv)
extends ThreadSafeRpcEndpoint with Logging {
+ private val epochBacklogQueueSize =
+ session.sqlContext.conf.continuousStreamingEpochBacklogQueueSize
+
private var queryWritesStopped: Boolean = false
private var numReaderPartitions: Int = _
@@ -212,6 +215,7 @@ private[continuous] class EpochCoordinator(
if (!partitionCommits.isDefinedAt((epoch, partitionId))) {
partitionCommits.put((epoch, partitionId), message)
resolveCommitsAtEpoch(epoch)
+ checkProcessingQueueBoundaries()
}
case ReportPartitionOffset(partitionId, epoch, offset) =>
@@ -223,6 +227,22 @@ private[continuous] class EpochCoordinator(
query.addOffset(epoch, stream, thisEpochOffsets.toSeq)
resolveCommitsAtEpoch(epoch)
}
+ checkProcessingQueueBoundaries()
+ }
+
+ private def checkProcessingQueueBoundaries() = {
+ if (partitionOffsets.size > epochBacklogQueueSize) {
+ query.stopInNewThread(new IllegalStateException("Size of the partition
offset queue has " +
+ "exceeded its maximum"))
+ }
+ if (partitionCommits.size > epochBacklogQueueSize) {
+ query.stopInNewThread(new IllegalStateException("Size of the partition
commit queue has " +
+ "exceeded its maximum"))
+ }
+ if (epochsWaitingToBeCommitted.size > epochBacklogQueueSize) {
+ query.stopInNewThread(new IllegalStateException("Size of the epoch queue
has " +
+ "exceeded its maximum"))
+ }
}
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any,
Unit] = {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
index 344a8aa..d2e489a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
@@ -25,6 +25,7 @@ import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.execution.streaming.continuous._
import org.apache.spark.sql.execution.streaming.sources.ContinuousMemoryStream
import org.apache.spark.sql.functions._
+import
org.apache.spark.sql.internal.SQLConf.CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE
import org.apache.spark.sql.streaming.{StreamTest, Trigger}
import org.apache.spark.sql.test.TestSparkSession
@@ -343,3 +344,33 @@ class ContinuousMetaSuite extends ContinuousSuiteBase {
}
}
}
+
+class ContinuousEpochBacklogSuite extends ContinuousSuiteBase {
+ import testImplicits._
+
+ override protected def createSparkSession = new TestSparkSession(
+ new SparkContext(
+ "local[1]",
+ "continuous-stream-test-sql-context",
+ sparkConf.set("spark.sql.testkey", "true")))
+
+ // This test forces the backlog to overflow by not standing up enough
executors for the query
+ // to make progress.
+ test("epoch backlog overflow") {
+ withSQLConf((CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE.key, "10")) {
+ val df = spark.readStream
+ .format("rate")
+ .option("numPartitions", "2")
+ .option("rowsPerSecond", "500")
+ .load()
+ .select('value)
+
+ testStream(df, useV2Sink = true)(
+ StartStream(Trigger.Continuous(1)),
+ ExpectFailure[IllegalStateException] { e =>
+ e.getMessage.contains("queue has exceeded its maximum")
+ }
+ )
+ }
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/EpochCoordinatorSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/EpochCoordinatorSuite.scala
index f74285f..e3498db 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/EpochCoordinatorSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/EpochCoordinatorSuite.scala
@@ -17,9 +17,9 @@
package org.apache.spark.sql.streaming.continuous
+import org.mockito.{ArgumentCaptor, InOrder}
import org.mockito.ArgumentMatchers.{any, eq => eqTo}
-import org.mockito.InOrder
-import org.mockito.Mockito.{inOrder, never, verify}
+import org.mockito.Mockito._
import org.scalatest.BeforeAndAfterEach
import org.scalatest.mockito.MockitoSugar
@@ -27,6 +27,7 @@ import org.apache.spark._
import org.apache.spark.rpc.RpcEndpointRef
import org.apache.spark.sql.LocalSparkSession
import org.apache.spark.sql.execution.streaming.continuous._
+import
org.apache.spark.sql.internal.SQLConf.CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE
import org.apache.spark.sql.sources.v2.reader.streaming.{ContinuousStream,
PartitionOffset}
import org.apache.spark.sql.sources.v2.writer.WriterCommitMessage
import org.apache.spark.sql.sources.v2.writer.streaming.StreamingWrite
@@ -43,6 +44,7 @@ class EpochCoordinatorSuite
private var writeSupport: StreamingWrite = _
private var query: ContinuousExecution = _
private var orderVerifier: InOrder = _
+ private val epochBacklogQueueSize = 10
override def beforeEach(): Unit = {
val stream = mock[ContinuousStream]
@@ -50,7 +52,11 @@ class EpochCoordinatorSuite
query = mock[ContinuousExecution]
orderVerifier = inOrder(writeSupport, query)
- spark = new TestSparkSession()
+ spark = new TestSparkSession(
+ new SparkContext(
+ "local[2]", "test-sql-context",
+ new SparkConf().set("spark.sql.testkey", "true")
+ .set(CONTINUOUS_STREAMING_EPOCH_BACKLOG_QUEUE_SIZE,
epochBacklogQueueSize)))
epochCoordinator
= EpochCoordinatorRef.create(writeSupport, stream, query, "test", 1,
spark, SparkEnv.get)
@@ -186,6 +192,66 @@ class EpochCoordinatorSuite
verifyCommitsInOrderOf(List(1, 2, 3, 4, 5))
}
+ test("several epochs, max epoch backlog reached by partitionOffsets") {
+ setWriterPartitions(1)
+ setReaderPartitions(1)
+
+ reportPartitionOffset(0, 1)
+ // Commit messages not arriving
+ for (i <- 2 to epochBacklogQueueSize + 1) {
+ reportPartitionOffset(0, i)
+ }
+
+ makeSynchronousCall()
+
+ for (i <- 1 to epochBacklogQueueSize + 1) {
+ verifyNoCommitFor(i)
+ }
+ verifyStoppedWithException("Size of the partition offset queue has
exceeded its maximum")
+ }
+
+ test("several epochs, max epoch backlog reached by partitionCommits") {
+ setWriterPartitions(1)
+ setReaderPartitions(1)
+
+ commitPartitionEpoch(0, 1)
+ // Offset messages not arriving
+ for (i <- 2 to epochBacklogQueueSize + 1) {
+ commitPartitionEpoch(0, i)
+ }
+
+ makeSynchronousCall()
+
+ for (i <- 1 to epochBacklogQueueSize + 1) {
+ verifyNoCommitFor(i)
+ }
+ verifyStoppedWithException("Size of the partition commit queue has
exceeded its maximum")
+ }
+
+ test("several epochs, max epoch backlog reached by
epochsWaitingToBeCommitted") {
+ setWriterPartitions(2)
+ setReaderPartitions(2)
+
+ commitPartitionEpoch(0, 1)
+ reportPartitionOffset(0, 1)
+
+ // For partition 2 epoch 1 messages never arriving
+ // +2 because the first epoch not yet arrived
+ for (i <- 2 to epochBacklogQueueSize + 2) {
+ commitPartitionEpoch(0, i)
+ reportPartitionOffset(0, i)
+ commitPartitionEpoch(1, i)
+ reportPartitionOffset(1, i)
+ }
+
+ makeSynchronousCall()
+
+ for (i <- 1 to epochBacklogQueueSize + 2) {
+ verifyNoCommitFor(i)
+ }
+ verifyStoppedWithException("Size of the epoch queue has exceeded its
maximum")
+ }
+
private def setWriterPartitions(numPartitions: Int): Unit = {
epochCoordinator.askSync[Unit](SetWriterPartitions(numPartitions))
}
@@ -221,4 +287,13 @@ class EpochCoordinatorSuite
private def verifyCommitsInOrderOf(epochs: Seq[Long]): Unit = {
epochs.foreach(verifyCommit)
}
+
+ private def verifyStoppedWithException(msg: String): Unit = {
+ val exceptionCaptor = ArgumentCaptor.forClass(classOf[Throwable]);
+ verify(query, atLeastOnce()).stopInNewThread(exceptionCaptor.capture())
+
+ import scala.collection.JavaConverters._
+ val throwable = exceptionCaptor.getAllValues.asScala.find(_.getMessage ===
msg)
+ assert(throwable != null, "Stream stopped with an exception but expected
message is missing")
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
