You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/06/09 16:00:02 UTC
[GitHub] [spark] yaooqinn commented on a change in pull request #28751: [SPARK-31926][SQL] Fix concurrency issue for ThriftCLIService to getPortNumber
yaooqinn commented on a change in pull request #28751:
URL: https://github.com/apache/spark/pull/28751#discussion_r436649612
##########
File path: sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SharedThriftServer.scala
##########
@@ -74,18 +86,31 @@ trait SharedThriftServer extends SharedSparkSession {
// Set the HIVE_SERVER2_THRIFT_PORT to 0, so it could randomly pick any free port to use.
// It's much more robust than set a random port generated by ourselves ahead
sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_PORT.varname, "0")
- hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
- hiveServer2.getServices.asScala.foreach {
- case t: ThriftCLIService if t.getPortNumber != 0 =>
- serverPort = t.getPortNumber
- logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
- case _ =>
- }
+ // Set the HIVE_SERVER2_THRIFT_HTTP_PORT to 0, so it could randomly pick any free port to use.
+ // It's much more robust than set a random port generated by ourselves ahead
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_HTTP_PORT.varname, "0")
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_TRANSPORT_MODE.varname, mode.toString)
+
+ try {
+ hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
+ hiveServer2.getServices.asScala.foreach {
+ case t: ThriftCLIService if t.getPortNumber != 0 =>
+ serverPort = t.getPortNumber
+ logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
Review comment:
Before this fix, yes. The port binding is in another background thread.
##########
File path: sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SharedThriftServer.scala
##########
@@ -74,18 +86,31 @@ trait SharedThriftServer extends SharedSparkSession {
// Set the HIVE_SERVER2_THRIFT_PORT to 0, so it could randomly pick any free port to use.
// It's much more robust than set a random port generated by ourselves ahead
sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_PORT.varname, "0")
- hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
- hiveServer2.getServices.asScala.foreach {
- case t: ThriftCLIService if t.getPortNumber != 0 =>
- serverPort = t.getPortNumber
- logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
- case _ =>
- }
+ // Set the HIVE_SERVER2_THRIFT_HTTP_PORT to 0, so it could randomly pick any free port to use.
+ // It's much more robust than set a random port generated by ourselves ahead
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_HTTP_PORT.varname, "0")
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_TRANSPORT_MODE.varname, mode.toString)
+
+ try {
+ hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
+ hiveServer2.getServices.asScala.foreach {
+ case t: ThriftCLIService if t.getPortNumber != 0 =>
+ serverPort = t.getPortNumber
+ logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
Review comment:
With https://github.com/apache/spark/pull/28751/files#diff-7610697b4f8f1bc4842c77e50807914cR178 and its implementations, the port binding is done in the same thread where we call `getPortNumber` later.
##########
File path: sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SharedThriftServer.scala
##########
@@ -74,18 +86,31 @@ trait SharedThriftServer extends SharedSparkSession {
// Set the HIVE_SERVER2_THRIFT_PORT to 0, so it could randomly pick any free port to use.
// It's much more robust than set a random port generated by ourselves ahead
sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_PORT.varname, "0")
- hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
- hiveServer2.getServices.asScala.foreach {
- case t: ThriftCLIService if t.getPortNumber != 0 =>
- serverPort = t.getPortNumber
- logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
- case _ =>
- }
+ // Set the HIVE_SERVER2_THRIFT_HTTP_PORT to 0, so it could randomly pick any free port to use.
+ // It's much more robust than set a random port generated by ourselves ahead
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_HTTP_PORT.varname, "0")
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_TRANSPORT_MODE.varname, mode.toString)
+
+ try {
+ hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
+ hiveServer2.getServices.asScala.foreach {
+ case t: ThriftCLIService if t.getPortNumber != 0 =>
+ serverPort = t.getPortNumber
+ logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
Review comment:
https://github.com/apache/spark/pull/28651#discussion_r435932239 there is a discussion with @juliuszsompolski before
##########
File path: sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SharedThriftServer.scala
##########
@@ -74,18 +86,31 @@ trait SharedThriftServer extends SharedSparkSession {
// Set the HIVE_SERVER2_THRIFT_PORT to 0, so it could randomly pick any free port to use.
// It's much more robust than set a random port generated by ourselves ahead
sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_PORT.varname, "0")
- hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
- hiveServer2.getServices.asScala.foreach {
- case t: ThriftCLIService if t.getPortNumber != 0 =>
- serverPort = t.getPortNumber
- logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
- case _ =>
- }
+ // Set the HIVE_SERVER2_THRIFT_HTTP_PORT to 0, so it could randomly pick any free port to use.
+ // It's much more robust than set a random port generated by ourselves ahead
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_HTTP_PORT.varname, "0")
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_TRANSPORT_MODE.varname, mode.toString)
+
+ try {
+ hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
+ hiveServer2.getServices.asScala.foreach {
+ case t: ThriftCLIService if t.getPortNumber != 0 =>
+ serverPort = t.getPortNumber
+ logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
Review comment:
https://github.com/apache/spark/pull/28651#discussion_r435932239 . there was a discussion with @juliuszsompolski before
##########
File path: sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SharedThriftServer.scala
##########
@@ -74,18 +86,31 @@ trait SharedThriftServer extends SharedSparkSession {
// Set the HIVE_SERVER2_THRIFT_PORT to 0, so it could randomly pick any free port to use.
// It's much more robust than set a random port generated by ourselves ahead
sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_PORT.varname, "0")
- hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
- hiveServer2.getServices.asScala.foreach {
- case t: ThriftCLIService if t.getPortNumber != 0 =>
- serverPort = t.getPortNumber
- logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
- case _ =>
- }
+ // Set the HIVE_SERVER2_THRIFT_HTTP_PORT to 0, so it could randomly pick any free port to use.
+ // It's much more robust than set a random port generated by ourselves ahead
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_THRIFT_HTTP_PORT.varname, "0")
+ sqlContext.setConf(ConfVars.HIVE_SERVER2_TRANSPORT_MODE.varname, mode.toString)
+
+ try {
+ hiveServer2 = HiveThriftServer2.startWithContext(sqlContext)
+ hiveServer2.getServices.asScala.foreach {
+ case t: ThriftCLIService if t.getPortNumber != 0 =>
+ serverPort = t.getPortNumber
+ logInfo(s"Started HiveThriftServer2: port=$serverPort, attempt=$attempt")
Review comment:
Take `ThriftBinaryCLIService` for an example
Before:
we do `TThreadPoolServer` initialization and `serve` in the same `run` function of the background thread. Then if we call getPortNumber right after `startWithContext`, concurrency issue will occur. The `portNum` may not reset yet when we call.
After:
we do `TThreadPoolServer` initialization in the current thread and do `serve` in the `run` function of the background thread.
##########
File path: sql/hive-thriftserver/v1.2/src/main/java/org/apache/hive/service/cli/thrift/ThriftBinaryCLIService.java
##########
@@ -45,13 +46,13 @@ public ThriftBinaryCLIService(CLIService cliService) {
}
@Override
- public void run() {
+ protected void initializeServer() {
try {
// Server thread pool
String threadPoolName = "HiveServer2-Handler-Pool";
ExecutorService executorService = new ThreadPoolExecutor(minWorkerThreads, maxWorkerThreads,
- workerKeepAliveTime, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
- new ThreadFactoryWithGarbageCleanup(threadPoolName));
+ workerKeepAliveTime, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
+ new ThreadFactoryWithGarbageCleanup(threadPoolName));
Review comment:
OK
##########
File path: sql/hive-thriftserver/v1.2/src/main/java/org/apache/hive/service/cli/thrift/ThriftBinaryCLIService.java
##########
@@ -97,14 +98,20 @@ public void run() {
// TCP Server
server = new TThreadPoolServer(sargs);
server.setServerEventHandler(serverEventHandler);
- String msg = "Starting " + ThriftBinaryCLIService.class.getSimpleName() + " on port "
- + serverSocket.getServerSocket().getLocalPort() + " with " + minWorkerThreads + "..." + maxWorkerThreads + " worker threads";
+ String msg = "Starting " + getName() + " on port " + portNum + " with " + minWorkerThreads +
+ "..." + maxWorkerThreads + " worker threads";
Review comment:
unnecessary change, you are right. reverted
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org