Revert "[SPARK-23785][LAUNCHER] LauncherBackend doesn't check state o…

…f connection before setting state"

This reverts commit 0bfbcaf.

core/src/main/scala/org/apache/spark/launcher/LauncherBackend.scala

@@ -67,13 +67,13 @@ private[spark] abstract class LauncherBackend {
   }
 
   def setAppId(appId: String): Unit = {
-    if (connection != null && isConnected) {
+    if (connection != null) {
       connection.send(new SetAppId(appId))
     }
   }
 
   def setState(state: SparkAppHandle.State): Unit = {
-    if (connection != null && isConnected && lastState != state) {
+    if (connection != null && lastState != state) {
       connection.send(new SetState(state))
       lastState = state
     }
@@ -114,10 +114,10 @@ private[spark] abstract class LauncherBackend {
 
     override def close(): Unit = {
       try {
-        _isConnected = false
         super.close()
       } finally {
         onDisconnected()
+        _isConnected = false
       }
     }
 

launcher/src/test/java/org/apache/spark/launcher/LauncherServerSuite.java

@@ -185,26 +185,6 @@ public void testStreamFiltering() throws Exception {
     }
   }
 
-  @Test
-  public void testAppHandleDisconnect() throws Exception {
-    LauncherServer server = LauncherServer.getOrCreateServer();
-    ChildProcAppHandle handle = new ChildProcAppHandle(server);
-    String secret = server.registerHandle(handle);
-
-    TestClient client = null;
-    try {
-      Socket s = new Socket(InetAddress.getLoopbackAddress(), server.getPort());
-      client = new TestClient(s);
-      client.send(new Hello(secret, "1.4.0"));
-      handle.disconnect();
-      waitForError(client, secret);
-    } finally {
-      handle.kill();
-      close(client);
-      client.clientThread.join();
-    }
-  }
-
   private void close(Closeable c) {
     if (c != null) {
       try {