Ngone51 commented on a change in pull request #28911:
URL: https://github.com/apache/spark/pull/28911#discussion_r457876717
##########
File path:
core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala
##########
@@ -194,6 +196,45 @@ private[spark] class NettyBlockTransferService(
result.future
}
+ override def getHostLocalDirs(
+ host: String,
+ port: Int,
+ execIds: Array[String],
+ hostLocalDirsCompletable: CompletableFuture[util.Map[String,
Array[String]]]): Unit = {
+ val getLocalDirsMessage = new GetLocalDirsForExecutors(appId, execIds)
+ try {
+ val client = clientFactory.createClient(host, port)
+ client.sendRpc(getLocalDirsMessage.toByteBuffer, new
RpcResponseCallback() {
+ override def onSuccess(response: ByteBuffer): Unit = {
+ try {
+ val msgObj = BlockTransferMessage.Decoder.fromByteBuffer(response)
+ hostLocalDirsCompletable.complete(
+ msgObj.asInstanceOf[LocalDirsForExecutors].getLocalDirsByExec)
+ } catch {
+ case t: Throwable =>
+ logWarning(s"Error trying to get the host local dirs for
executor ${execIds.head}",
+ t.getCause)
+ hostLocalDirsCompletable.completeExceptionally(t)
+ } finally {
+ client.close()
+ }
+ }
+
+ override def onFailure(t: Throwable): Unit = {
+ logWarning(s"Error trying to get the host local dirs for executor
${execIds.head}",
+ t.getCause)
+ hostLocalDirsCompletable.completeExceptionally(t)
+ client.close()
+ }
+ })
+ } catch {
+ case e: IOException =>
Review comment:
Yes. Good idea. I've did the refactor. Please take a look.
----------------------------------------------------------------
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]
