juliuszsompolski commented on a change in pull request #28912:
URL: https://github.com/apache/spark/pull/28912#discussion_r448285230
##########
File path:
sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperationSuite.scala
##########
@@ -42,4 +57,65 @@ class SparkExecuteStatementOperationSuite extends
SparkFunSuite {
assert(columns.get(1).getType().getName == "INT")
assert(columns.get(1).getComment() == "")
}
+
+ Seq(
+ (OperationState.CANCELED, (_: SparkExecuteStatementOperation).cancel()),
+ (OperationState.CLOSED, (_: SparkExecuteStatementOperation).close())
+ ).foreach { case (finalState, transition) =>
+ test("SPARK-32057 SparkExecuteStatementOperation should not transiently
become ERROR " +
+ s"before being set to $finalState") {
+ val hiveSession = new
HiveSessionImpl(ThriftserverShimUtils.testedProtocolVersions.head,
+ "username", "password", new HiveConf, "ip address")
+ hiveSession.open(new util.HashMap)
+
+ HiveThriftServer2.eventManager =
mock(classOf[HiveThriftServer2EventManager])
+
+ val spySqlContext = spy(sqlContext)
+
+ // When cancel() is called on the operation, cleanup causes an exception
to be thrown inside
+ // of execute(). This should not cause the state to become ERROR. The
exception here will be
+ // triggered in our custom cleanup().
+ val signal = new Semaphore(0)
+ val dataFrame = mock(classOf[DataFrame], RETURNS_DEEP_STUBS)
+ when(dataFrame.collect()).thenAnswer((_: InvocationOnMock) => {
+ signal.acquire()
+ throw new RuntimeException("Operation was cancelled by test cleanup.")
+ })
+ val statement = "stmt"
+ doReturn(dataFrame, Nil: _*).when(spySqlContext).sql(statement)
+
+ val executeStatementOperation = new
MySparkExecuteStatementOperation(spySqlContext,
+ hiveSession, statement, signal, finalState)
+
+ val run = new Thread() {
+ override def run(): Unit = executeStatementOperation.runInternal()
+ }
+ assert(executeStatementOperation.getStatus.getState ===
OperationState.INITIALIZED)
+ run.start()
+ eventually(timeout(5.seconds)) {
+ assert(executeStatementOperation.getStatus.getState ===
OperationState.RUNNING)
+ }
+ transition(executeStatementOperation)
+ run.join()
+ assert(executeStatementOperation.getStatus.getState === finalState)
+ }
+ }
+ private class MySparkExecuteStatementOperation(
+ sqlContext: SQLContext,
+ hiveSession: HiveSession,
+ statement: String,
+ signal: Semaphore,
+ finalState: OperationState)
+ extends SparkExecuteStatementOperation(sqlContext, hiveSession, statement,
+ new util.HashMap, false) {
+
+ override def cleanup(): Unit = {
+ super.cleanup()
+ signal.release()
+ // Allow time for the exception to propagate.
+ Thread.sleep(1000)
Review comment:
@alismess-db I was wondering about hte same place before as well.
Let's maybe add a comment explaining it like e.g.
`// At this point, operation should already be in finalState (set by either
close() or cancel()). We want to check if it stays in finalState after the
exception thrown from after releasing the semaphore propagates`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
