maropu commented on a change in pull request #28991:
URL: https://github.com/apache/spark/pull/28991#discussion_r449442618
##########
File path:
sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala
##########
@@ -874,6 +874,16 @@ class HiveThriftBinaryServerSuite extends
HiveThriftJdbcTest {
assert(rs.getString(1) === expected.toString)
}
}
+
+ test("SPARK-26533: Support query auto timeout cancel on thriftserver") {
+ withJdbcStatement() { statement =>
+ statement.setQueryTimeout(1)
Review comment:
Could you test the cases `0` and `-1`? They mean no limit?
```
/**
...
* @param seconds the new query timeout limit in seconds; zero means
* there is no limit
* @exception SQLException if a database access error occurs,
* this method is called on a closed <code>Statement</code>
* or the condition {@code seconds >= 0} is not satisfied
* @see #getQueryTimeout
*/
void setQueryTimeout(int seconds) throws SQLException;
```
##########
File path:
sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
##########
@@ -204,6 +205,13 @@ private[hive] class SparkExecuteStatementOperation(
parentSession.getUsername)
setHasResultSet(true) // avoid no resultset for async run
+ if(queryTimeout > 0) {
+ Executors.newSingleThreadScheduledExecutor
+ .schedule(new Runnable {
Review comment:
nit format:
```
Executors.newSingleThreadScheduledExecutor.schedule(new Runnable {
override def run(): Unit = timeoutCancel()
}, queryTimeout, TimeUnit.SECONDS)
```
##########
File path:
sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
##########
@@ -349,6 +357,17 @@ private[hive] class SparkExecuteStatementOperation(
}
}
+ def timeoutCancel(): Unit = {
Review comment:
Could we inline this method in the line 211?
##########
File path:
sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
##########
@@ -349,6 +357,17 @@ private[hive] class SparkExecuteStatementOperation(
}
}
+ def timeoutCancel(): Unit = {
+ synchronized {
+ if (!getStatus.getState.isTerminal) {
+ logInfo(s"Timeout and Cancel query with $statementId ")
+ cleanup()
+ setState(OperationState.TIMEDOUT)
Review comment:
nit: `logInfo` is located just before
`HiveThriftServer2.eventManager.onXXX`?
```
cleanup()
setState(OperationState.TIMEDOUT)
logInfo(s"Timeout and Cancel query with $statementId ")
HiveThriftServer2.eventManager.onStatementCanceled(statementId)
```
https://github.com/apache/spark/blob/42f01e314b4874236544cc8b94bef766269385ee/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkOperation.scala#L51-L52
##########
File path:
sql/hive-thriftserver/v1.2/src/main/java/org/apache/hive/service/cli/OperationState.java
##########
@@ -32,7 +32,8 @@
CLOSED(TOperationState.CLOSED_STATE, true),
ERROR(TOperationState.ERROR_STATE, true),
UNKNOWN(TOperationState.UKNOWN_STATE, false),
- PENDING(TOperationState.PENDING_STATE, false);
+ PENDING(TOperationState.PENDING_STATE, false),
+ TIMEDOUT(TOperationState.CANCELED_STATE, true); //do not want to change
TOperationState in hive 1.2
Review comment:
What does `//do not want to change TOperationState in hive 1.2` means?
----------------------------------------------------------------
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]
