Skip to content

[SPARK-23747][Structured Streaming] Add EpochCoordinator unit tests #20983

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
spaced4ndy wants to merge 5 commits into from
Closed

[SPARK-23747][Structured Streaming] Add EpochCoordinator unit tests #20983

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b5cd066
[SPARK-23747][Structured Streaming] Added EpochCoordinator unit tests
Apr 5, 2018
284795b
Passing raw numbers for epochs as parameters
Apr 10, 2018
7d85cf2
Added tests
Apr 10, 2018
89a1e11
Renamed method
Apr 11, 2018
8fa609c
Replaced SharedSparkSession with LocalSparkSession
Apr 13, 2018
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
Replaced SharedSparkSession with LocalSparkSession
  • Loading branch information
Efim Poberezkin committed Apr 13, 2018
commit 8fa609cd8ad6130aa16b9bf624fe5b5e0f5ef256
11 changes: 5 additions & 6 deletions ...core/src/test/scala/org/apache/spark/sql/streaming/continuous/EpochCoordinatorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,16 @@ import org.scalatest.mockito.MockitoSugar

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.sources.v2.reader.streaming.{ContinuousReader, PartitionOffset}
import org.apache.spark.sql.sources.v2.writer.WriterCommitMessage
import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.test.TestSparkSession

class EpochCoordinatorSuite
extends SparkFunSuite
with SharedSparkSession
with LocalSparkSession
with MockitoSugar
with BeforeAndAfterEach {

Expand All @@ -49,14 +50,12 @@ class EpochCoordinatorSuite
query = mock[ContinuousExecution]
orderVerifier = inOrder(writer, query)

spark = new TestSparkSession()

epochCoordinator
= EpochCoordinatorRef.create(writer, reader, query, "test", 1, spark, SparkEnv.get)
}

override def afterEach(): Unit = {
SparkEnv.get.rpcEnv.stop(epochCoordinator)
}

test("single epoch") {
setWriterPartitions(3)
setReaderPartitions(2)
Expand Down