Explicitly mark the client as timed out.

Since ctx.close() is asynchronous, this ensures that threads checking
for the client being alive get the right result.

Author: Marcelo Vanzin

network/common/src/main/java/org/apache/spark/network/client/TransportClient.java

@@ -73,18 +73,20 @@ public class TransportClient implements Closeable {
   private final Channel channel;
   private final TransportResponseHandler handler;
   @Nullable private String clientId;
+  private volatile boolean timedOut;
 
   public TransportClient(Channel channel, TransportResponseHandler handler) {
     this.channel = Preconditions.checkNotNull(channel);
     this.handler = Preconditions.checkNotNull(handler);
+    this.timedOut = false;
   }
 
   public Channel getChannel() {
     return channel;
   }
 
   public boolean isActive() {
-    return channel.isOpen() || channel.isActive();
+    return !timedOut && (channel.isOpen() || channel.isActive());
   }
 
   public SocketAddress getSocketAddress() {
@@ -263,6 +265,11 @@ public void onFailure(Throwable e) {
     }
   }
 
+  /** Mark this channel as having timed out. */
+  public void timeOut() {
+    this.timedOut = true;
+  }
+
   @Override
   public void close() {
     // close is a local operation and should finish with milliseconds; timeout just to be safe

network/common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java

@@ -128,9 +128,11 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
             logger.error("Connection to {} has been quiet for {} ms while there are outstanding " +
               "requests. Assuming connection is dead; please adjust spark.network.timeout if this " +
               "is wrong.", address, requestTimeoutNs / 1000 / 1000);
+            client.timeOut();
             ctx.close();
           } else if (closeIdleConnections) {
             // While CloseIdleConnections is enable, we also close idle connection
+            client.timeOut();
             ctx.close();
           }
         }