Skip to content

Commit 783fc16

Browse files
yaooqinnulysses-you
yaooqinn
authored and
ulysses-you
committed
[KYUUBI #1766] [TEST] Remove SessionManagerSuite to reduce duplicated tests
<!-- Thanks for sending a pull request! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html 2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'. 3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'. --> ### _Why are the changes needed?_ <!-- Please clarify why the changes are needed. For instance, 1. If you add a feature, you can talk about the use case of it. 2. If you fix a bug, you can clarify why it is a bug. --> SessionManagerSuite re-runs all tests in the super class, it's unnecessary. ### _How was this patch tested?_ - [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible - [ ] Add screenshots for manual tests if appropriate - [x] [Run test](https://kyuubi.readthedocs.io/en/latest/develop_tools/testing.html#running-tests) locally before make a pull request Closes #1766 from yaooqinn/test2. Closes #1766 2825073 [Kent Yao] [TEST] Remove SessionManagerSuite to reduce duplicated tests Authored-by: Kent Yao <yao@apache.org> Signed-off-by: ulysses-you <ulyssesyou@apache.org>
1 parent 3881469 commit 783fc16

File tree

4 files changed

+167
-200
lines changed

4 files changed

+167
-200
lines changed

‎kyuubi-common/src/test/scala/org/apache/kyuubi/operation/HiveJDBCTestHelper.scala

Lines changed: 5 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -19,17 +19,12 @@ package org.apache.kyuubi.operation
1919

2020
import java.sql.ResultSet
2121

22-
import scala.collection.JavaConverters._
23-
2422
import org.apache.hive.service.rpc.thrift._
2523
import org.apache.hive.service.rpc.thrift.TCLIService.Iface
2624
import org.apache.hive.service.rpc.thrift.TOperationState._
27-
import org.apache.thrift.protocol.TBinaryProtocol
28-
import org.apache.thrift.transport.TSocket
2925
import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
3026

3127
import org.apache.kyuubi.Utils
32-
import org.apache.kyuubi.service.authentication.PlainSASLHelper
3328

3429
trait HiveJDBCTestHelper extends JDBCTestHelper {
3530

@@ -81,43 +76,13 @@ trait HiveJDBCTestHelper extends JDBCTestHelper {
8176
jdbcUrl + sessionConfStr + jdbcConfStr + jdbcVarsStr
8277
}
8378

84-
def withThriftClient(f: TCLIService.Iface => Unit): Unit = {
85-
val hostAndPort = jdbcUrl.stripPrefix("jdbc:hive2://").split("/;").head.split(":")
86-
val host = hostAndPort.head
87-
val port = hostAndPort(1).toInt
88-
val socket = new TSocket(host, port)
89-
val transport = PlainSASLHelper.getPlainTransport(Utils.currentUser, password, socket)
90-
91-
val protocol = new TBinaryProtocol(transport)
92-
val client = new TCLIService.Client(protocol)
93-
transport.open()
94-
try {
95-
f(client)
96-
} finally {
97-
socket.close()
98-
}
79+
def withThriftClient[T](f: TCLIService.Iface => T): T = {
80+
TClientTestUtils.withThriftClient(jdbcUrl.stripPrefix("jdbc:hive2://").split("/;").head)(f)
9981
}
10082

101-
def withSessionHandle(f: (TCLIService.Iface, TSessionHandle) => Unit): Unit = {
102-
withThriftClient { client =>
103-
val req = new TOpenSessionReq()
104-
req.setUsername(user)
105-
req.setPassword(password)
106-
req.setConfiguration(_sessionConfigs.asJava)
107-
val resp = client.OpenSession(req)
108-
val handle = resp.getSessionHandle
109-
110-
try {
111-
f(client, handle)
112-
} finally {
113-
val tCloseSessionReq = new TCloseSessionReq(handle)
114-
try {
115-
client.CloseSession(tCloseSessionReq)
116-
} catch {
117-
case e: Exception => error(s"Failed to close $handle", e)
118-
}
119-
}
120-
}
83+
def withSessionHandle[T](f: (TCLIService.Iface, TSessionHandle) => T): T = {
84+
val hostAndPort = jdbcUrl.stripPrefix("jdbc:hive2://").split("/;").head
85+
TClientTestUtils.withSessionHandle(hostAndPort, sessionConfigs)(f)
12186
}
12287

12388
def checkGetSchemas(rs: ResultSet, dbNames: Seq[String], catalogName: String = ""): Unit = {

‎kyuubi-common/src/test/scala/org/apache/kyuubi/operation/TClientTestUtils.scala

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.kyuubi.operation
19+
20+
import scala.collection.JavaConverters._
21+
22+
import org.apache.hive.service.rpc.thrift.{TCLIService, TCloseSessionReq, TOpenSessionReq, TSessionHandle}
23+
import org.apache.hive.service.rpc.thrift.TCLIService.Iface
24+
import org.apache.thrift.protocol.TBinaryProtocol
25+
import org.apache.thrift.transport.TSocket
26+
27+
import org.apache.kyuubi.{Logging, Utils}
28+
import org.apache.kyuubi.service.FrontendService
29+
import org.apache.kyuubi.service.authentication.PlainSASLHelper
30+
31+
object TClientTestUtils extends Logging {
32+
33+
def withThriftClient[T](url: String)(f: Iface => T): T = {
34+
val hostport = url.split(':')
35+
val socket = new TSocket(hostport.head, hostport.last.toInt)
36+
val transport = PlainSASLHelper.getPlainTransport(Utils.currentUser, "anonymous", socket)
37+
val protocol = new TBinaryProtocol(transport)
38+
val client = new TCLIService.Client(protocol)
39+
transport.open()
40+
try {
41+
f(client)
42+
} finally {
43+
socket.close()
44+
}
45+
}
46+
47+
/**
48+
* s shall be [[TFrontendService]]
49+
*/
50+
def withThriftClient[T](s: FrontendService)(f: Iface => T): T = {
51+
withThriftClient(s.connectionUrl)(f)
52+
}
53+
54+
def withSessionHandle[T](url: String, configs: Map[String, String])(
55+
f: (TCLIService.Iface, TSessionHandle) => T): T = {
56+
withThriftClient(url) { client =>
57+
val req = new TOpenSessionReq()
58+
req.setUsername(Utils.currentUser)
59+
req.setPassword("anonymous")
60+
req.setConfiguration(configs.asJava)
61+
val resp = client.OpenSession(req)
62+
val handle = resp.getSessionHandle
63+
try {
64+
f(client, handle)
65+
} finally {
66+
val tCloseSessionReq = new TCloseSessionReq(handle)
67+
try {
68+
client.CloseSession(tCloseSessionReq)
69+
} catch {
70+
case e: Exception => error(s"Failed to close $handle", e)
71+
}
72+
}
73+
}
74+
}
75+
}

‎kyuubi-common/src/test/scala/org/apache/kyuubi/service/ThriftFrontendServiceSuite.scala renamed to ‎kyuubi-common/src/test/scala/org/apache/kyuubi/service/TFrontendServiceSuite.scala

Lines changed: 87 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -17,31 +17,35 @@
1717

1818
package org.apache.kyuubi.service
1919

20-
import java.util
20+
import java.time.Duration
2121

2222
import scala.collection.JavaConverters._
2323

2424
import org.apache.hive.service.rpc.thrift._
25-
import org.apache.thrift.protocol.TBinaryProtocol
26-
import org.apache.thrift.transport.TSocket
25+
import org.scalatest.time._
2726

2827
import org.apache.kyuubi.{KyuubiFunSuite, KyuubiSQLException, Utils}
2928
import org.apache.kyuubi.config.KyuubiConf
3029
import org.apache.kyuubi.config.KyuubiConf.{FRONTEND_BIND_HOST, FRONTEND_CONNECTION_URL_USE_HOSTNAME, FRONTEND_THRIFT_BINARY_BIND_HOST, FRONTEND_THRIFT_BINARY_BIND_PORT}
31-
import org.apache.kyuubi.operation.{OperationHandle, OperationType}
30+
import org.apache.kyuubi.operation.{OperationHandle, OperationType, TClientTestUtils}
3231
import org.apache.kyuubi.service.TFrontendService.{FeServiceServerContext, SERVER_VERSION}
33-
import org.apache.kyuubi.service.authentication.PlainSASLHelper
34-
import org.apache.kyuubi.session.SessionHandle
32+
import org.apache.kyuubi.session.{AbstractSession, SessionHandle}
3533

36-
class ThriftFrontendServiceSuite extends KyuubiFunSuite {
34+
class TFrontendServiceSuite extends KyuubiFunSuite {
3735

3836
protected val server = new NoopTBinaryFrontendServer()
3937
protected val conf = KyuubiConf()
4038
.set(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
4139
.set("kyuubi.test.server.should.fail", "false")
42-
43-
val user: String = System.getProperty("user.name")
44-
val sessionConf: util.Map[String, String] = new util.HashMap()
40+
.set(KyuubiConf.SESSION_CHECK_INTERVAL, Duration.ofSeconds(5).toMillis)
41+
.set(KyuubiConf.SESSION_IDLE_TIMEOUT, Duration.ofSeconds(5).toMillis)
42+
.set(KyuubiConf.OPERATION_IDLE_TIMEOUT, Duration.ofSeconds(20).toMillis)
43+
.set(KyuubiConf.SESSION_CONF_RESTRICT_LIST, Seq("spark.*"))
44+
.set(KyuubiConf.SESSION_CONF_IGNORE_LIST, Seq("session.engine.*"))
45+
46+
private def withSessionHandle(f: (TCLIService.Iface, TSessionHandle) => Unit): Unit = {
47+
TClientTestUtils.withSessionHandle(server.frontendServices.head.connectionUrl, Map.empty)(f)
48+
}
4549

4650
override def beforeAll(): Unit = {
4751
server.initialize(conf)
@@ -54,45 +58,6 @@ class ThriftFrontendServiceSuite extends KyuubiFunSuite {
5458
super.afterAll()
5559
}
5660

57-
protected def withThriftClient(f: TCLIService.Iface => Unit): Unit = {
58-
val hostAndPort = server.frontendServices.head.connectionUrl.split(":")
59-
val host = hostAndPort.head
60-
val port = hostAndPort(1).toInt
61-
val socket = new TSocket(host, port)
62-
val transport = PlainSASLHelper.getPlainTransport(Utils.currentUser, "anonymous", socket)
63-
64-
val protocol = new TBinaryProtocol(transport)
65-
val client = new TCLIService.Client(protocol)
66-
transport.open()
67-
try {
68-
f(client)
69-
} finally {
70-
socket.close()
71-
}
72-
}
73-
74-
protected def withSessionHandle(f: (TCLIService.Iface, TSessionHandle) => Unit): Unit = {
75-
withThriftClient { client =>
76-
val req = new TOpenSessionReq()
77-
req.setUsername(user)
78-
req.setPassword("anonymous")
79-
req.setConfiguration(sessionConf)
80-
val resp = client.OpenSession(req)
81-
val handle = resp.getSessionHandle
82-
83-
try {
84-
f(client, handle)
85-
} finally {
86-
val tCloseSessionReq = new TCloseSessionReq(handle)
87-
try {
88-
client.CloseSession(tCloseSessionReq)
89-
} catch {
90-
case e: Exception => error(s"Failed to close $handle", e)
91-
}
92-
}
93-
}
94-
}
95-
9661
private def checkOperationResult(
9762
client: TCLIService.Iface,
9863
handle: TOperationHandle): Unit = {
@@ -151,24 +116,26 @@ class ThriftFrontendServiceSuite extends KyuubiFunSuite {
151116
}
152117

153118
test("open session") {
154-
withThriftClient { client =>
155-
val req = new TOpenSessionReq()
156-
req.setUsername(user)
157-
req.setPassword("anonymous")
158-
val resp = client.OpenSession(req)
159-
val handle = resp.getSessionHandle
160-
assert(handle != null)
161-
assert(resp.getStatus.getStatusCode == TStatusCode.SUCCESS_STATUS)
162-
163-
req.setConfiguration(Map("kyuubi.test.should.fail" -> "true").asJava)
164-
val resp1 = client.OpenSession(req)
165-
assert(resp1.getSessionHandle === null)
166-
assert(resp1.getStatus.getStatusCode === TStatusCode.ERROR_STATUS)
167-
val cause = KyuubiSQLException.toCause(resp1.getStatus.getInfoMessages.asScala)
168-
assert(cause.isInstanceOf[KyuubiSQLException])
169-
assert(cause.getMessage === "Asked to fail")
119+
TClientTestUtils.withThriftClient(server.frontendServices.head) {
120+
client =>
121+
val req = new TOpenSessionReq()
122+
req.setUsername(Utils.currentUser)
123+
req.setPassword("anonymous")
124+
val resp = client.OpenSession(req)
125+
val handle = resp.getSessionHandle
126+
assert(handle != null)
127+
assert(resp.getStatus.getStatusCode == TStatusCode.SUCCESS_STATUS)
128+
129+
req.setConfiguration(Map("kyuubi.test.should.fail" -> "true").asJava)
130+
val resp1 = client.OpenSession(req)
131+
assert(resp1.getSessionHandle === null)
132+
assert(resp1.getStatus.getStatusCode === TStatusCode.ERROR_STATUS)
133+
val cause = KyuubiSQLException.toCause(resp1.getStatus.getInfoMessages.asScala)
134+
assert(cause.isInstanceOf[KyuubiSQLException])
135+
assert(cause.getMessage === "Asked to fail")
136+
137+
assert(resp1.getStatus.getErrorMessage === "Asked to fail")
170138

171-
assert(resp1.getStatus.getErrorMessage === "Asked to fail")
172139
}
173140
}
174141

@@ -513,4 +480,58 @@ class ThriftFrontendServiceSuite extends KyuubiFunSuite {
513480
"Delegation token is not supported")
514481
}
515482
}
483+
484+
test("close expired operations") {
485+
withSessionHandle { (client, handle) =>
486+
val req = new TCancelOperationReq()
487+
val req1 = new TGetSchemasReq(handle)
488+
val resp1 = client.GetSchemas(req1)
489+
490+
val sessionManager = server.backendService.sessionManager
491+
val session = sessionManager
492+
.getSession(SessionHandle(handle))
493+
.asInstanceOf[AbstractSession]
494+
var lastAccessTime = session.lastAccessTime
495+
assert(sessionManager.getOpenSessionCount == 1)
496+
assert(session.lastIdleTime > 0)
497+
498+
resp1.getOperationHandle
499+
req.setOperationHandle(resp1.getOperationHandle)
500+
val resp2 = client.CancelOperation(req)
501+
assert(resp2.getStatus.getStatusCode === TStatusCode.SUCCESS_STATUS)
502+
assert(sessionManager.getOpenSessionCount == 1)
503+
assert(session.lastIdleTime == 0)
504+
assert(lastAccessTime < session.lastAccessTime)
505+
lastAccessTime = session.lastAccessTime
506+
507+
eventually(timeout(Span(60, Seconds)), interval(Span(1, Seconds))) {
508+
assert(session.lastIdleTime > lastAccessTime)
509+
}
510+
511+
info("operation is terminated")
512+
assert(lastAccessTime == session.lastAccessTime)
513+
assert(sessionManager.getOpenSessionCount == 1)
514+
515+
eventually(timeout(Span(60, Seconds)), interval(Span(1, Seconds))) {
516+
assert(session.lastAccessTime > lastAccessTime)
517+
}
518+
assert(sessionManager.getOpenSessionCount == 0)
519+
}
520+
}
521+
522+
test("test validate and normalize config") {
523+
val sessionManager = server.backendService.sessionManager
524+
// test restrict
525+
intercept[KyuubiSQLException] {
526+
sessionManager.validateAndNormalizeConf(Map("spark.driver.memory" -> "2G"))
527+
}
528+
529+
// test ignore
530+
val conf = sessionManager.validateAndNormalizeConf(
531+
Map(
532+
"session.engine.spark.main.resource" -> "org.apahce.kyuubi.test",
533+
"session.check.interval" -> "10000"))
534+
assert(conf.size == 1)
535+
assert(conf("session.check.interval") == "10000")
536+
}
516537
}

0 commit comments

Comments
 (0)