Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20983#discussion_r180553748
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/EpochCoordinatorSuite.scala
---
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming.continuous
+
+import org.mockito.InOrder
+import org.mockito.Matchers.{any, eq => eqTo}
+import org.mockito.Mockito._
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.mockito.MockitoSugar
+
+import org.apache.spark._
+import org.apache.spark.rpc.RpcEndpointRef
+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
+
+class EpochCoordinatorSuite
+ extends SparkFunSuite
+ with SharedSparkSession
+ with MockitoSugar
+ with BeforeAndAfterEach {
+
+ private var epochCoordinator: RpcEndpointRef = _
+
+ private var writer: StreamWriter = _
+ private var query: ContinuousExecution = _
+ private var orderVerifier: InOrder = _
+
+ override def beforeEach(): Unit = {
+ val reader = mock[ContinuousReader]
+ writer = mock[StreamWriter]
+ query = mock[ContinuousExecution]
+ orderVerifier = inOrder(writer, query)
+
+ 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)
+
+ commitPartitionEpoch(0, 1)
+ commitPartitionEpoch(1, 1)
+ commitPartitionEpoch(2, 1)
+ reportPartitionOffset(0, 1)
+ reportPartitionOffset(1, 1)
+
+ // Here and in subsequent tests this is called to make a synchronous
call to EpochCoordinator
+ // so that mocks would have been acted upon by the time verification
happens
+ makeSynchronousCall()
+
+ verifyCommit(1)
+ }
+
+ test("single epoch, all but one writer partition has committed") {
+ setWriterPartitions(3)
+ setReaderPartitions(2)
+
+ commitPartitionEpoch(0, 1)
+ commitPartitionEpoch(1, 1)
+ reportPartitionOffset(0, 1)
+ reportPartitionOffset(1, 1)
+
+ makeSynchronousCall()
+
+ verifyCommitHasntHappened(1)
+ }
+
+ test("single epoch, all but one reader partition has reported an
offset") {
+ setWriterPartitions(3)
+ setReaderPartitions(2)
+
+ commitPartitionEpoch(0, 1)
+ commitPartitionEpoch(1, 1)
+ commitPartitionEpoch(2, 1)
+ reportPartitionOffset(0, 1)
+
+ makeSynchronousCall()
+
+ verifyCommitHasntHappened(1)
--- End diff --
+1
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
