[incubator-celeborn] branch branch-0.1 updated: [ISSUE-228][BUG] When CommitFile, worker endpoint may be closed by ReserveSlots failed, we should try reconnect endpoint when endpoint not active (#976)

[zh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhouky pushed a commit to branch branch-0.1
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/branch-0.1 by this push:
     new c89f6368 [ISSUE-228][BUG] When CommitFile, worker endpoint may be closed by ReserveSlots failed, we should try reconnect endpoint when endpoint not active (#976)
c89f6368 is described below

commit c89f6368c13f4f7c267ce27f1600b7d598c2aabd
Author: Keyong Zhou <zk...@alibaba-inc.com>
AuthorDate: Thu Nov 17 15:11:28 2022 +0800

    [ISSUE-228][BUG] When CommitFile, worker endpoint may be closed by ReserveSlots failed, we should try reconnect endpoint when endpoint not active (#976)
    
    * [ISSUE-228][BUG] When CommitFile, worker endpoint may be closed by ReserveSlots failed, we should try reconnect endpoint when endpoint not active (#231)
    
    * [ISSUE-228]Fix unexpected closed exceptions occurred while committing files. (#232)
    
    Co-authored-by: AngersZhuuuu <an...@gmail.com>
    Co-authored-by: Ethan Feng <fe...@alibaba-inc.com>
---
 .../com/aliyun/emr/rss/client/write/LifecycleManager.scala     | 10 ----------
 .../com/aliyun/emr/rss/common/rpc/netty/NettyRpcEnv.scala      |  6 +++---
 .../scala/com/aliyun/emr/rss/common/rpc/netty/Outbox.scala     |  2 +-
 3 files changed, 4 insertions(+), 14 deletions(-)

diff --git a/client/src/main/scala/com/aliyun/emr/rss/client/write/LifecycleManager.scala b/client/src/main/scala/com/aliyun/emr/rss/client/write/LifecycleManager.scala
index 42de5301..52319f0d 100644
--- a/client/src/main/scala/com/aliyun/emr/rss/client/write/LifecycleManager.scala
+++ b/client/src/main/scala/com/aliyun/emr/rss/client/write/LifecycleManager.scala
@@ -312,9 +312,6 @@ class LifecycleManager(appId: String, val conf: RssConf) extends RpcEndpoint wit
       try {
         workerInfo.endpoint = rpcEnv.setupEndpointRef(
           RpcAddress.apply(workerInfo.host, workerInfo.rpcPort), WORKER_EP)
-        workerInfo.endpoint.asInstanceOf[NettyRpcEndpointRef].client =
-          rpcEnv.asInstanceOf[NettyRpcEnv].clientFactory.createClient(workerInfo.host,
-            workerInfo.rpcPort)
       } catch {
         case t: Throwable =>
           logError(s"Init rpc client for $workerInfo failed", t)
@@ -866,13 +863,6 @@ class LifecycleManager(appId: String, val conf: RssConf) extends RpcEndpoint wit
             }
           })
         }
-        // remove failed slot from total slots, close transport client
-        if (workerInfo.endpoint != null) {
-          val transportClient = workerInfo.endpoint.asInstanceOf[NettyRpcEndpointRef].client
-          if (null != transportClient && transportClient.isActive) {
-            transportClient.close()
-          }
-        }
       })
 
       val newMapFunc = new util.function.Function[WorkerInfo,
diff --git a/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/NettyRpcEnv.scala b/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/NettyRpcEnv.scala
index d77be335..8487c737 100644
--- a/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/NettyRpcEnv.scala
+++ b/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/NettyRpcEnv.scala
@@ -18,7 +18,7 @@
 package com.aliyun.emr.rss.common.rpc.netty
 
 import java.io._
-import java.net.{InetSocketAddress}
+import java.net.InetSocketAddress
 import java.nio.ByteBuffer
 import java.util.concurrent._
 import java.util.concurrent.atomic.AtomicBoolean
@@ -26,7 +26,7 @@ import javax.annotation.Nullable
 
 import scala.concurrent.{Future, Promise}
 import scala.reflect.ClassTag
-import scala.util.{DynamicVariable, Failure, Success, Try}
+import scala.util.{DynamicVariable, Failure, Success}
 import scala.util.control.NonFatal
 
 import com.google.common.base.Throwables
@@ -139,7 +139,7 @@ class NettyRpcEnv(
   }
 
   private def postToOutbox(receiver: NettyRpcEndpointRef, message: OutboxMessage): Unit = {
-    if (receiver.client != null) {
+    if (receiver.client != null && receiver.client.isActive) {
       message.sendWith(receiver.client)
     } else {
       require(receiver.address != null,
diff --git a/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/Outbox.scala b/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/Outbox.scala
index 9f62c824..f3c0a569 100644
--- a/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/Outbox.scala
+++ b/common/src/main/scala/com/aliyun/emr/rss/common/rpc/netty/Outbox.scala
@@ -145,7 +145,7 @@ private[rss] class Outbox(nettyEnv: NettyRpcEnv, val address: RpcAddress) {
         // We are connecting to the remote address, so just exit
         return
       }
-      if (client == null) {
+      if (client == null || !client.isActive) {
         // There is no connect task but client is null, so we need to launch the connect task.
         launchConnectTask()
         return