Merge -r 947746:947747 from trunk to 0.21 branch. Fixes: HADOOP-6723.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.21@947756 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -1527,6 +1527,9 @@ Release 0.21.0 - Unreleased
     despite failure at any level. (Contributed by Ravi Gummadi and 
     Vinod Kumar Vavilapalli)
 
+    HADOOP-6723.  Unchecked exceptions thrown in IPC Connection should not
+    orphan clients.  (Todd Lipcon via tomwhite)
+
 Release 0.20.3 - Unreleased
 
   NEW FEATURES

src/java/org/apache/hadoop/ipc/Client.java

@@ -597,8 +597,16 @@ public class Client {
         LOG.debug(getName() + ": starting, having connections " 
             + connections.size());
 
-      while (waitForWork()) {//wait here for work - read or close connection
-        receiveResponse();
+      try {
+        while (waitForWork()) {//wait here for work - read or close connection
+          receiveResponse();
+        }
+      } catch (Throwable t) {
+        // This truly is unexpected, since we catch IOException in receiveResponse
+        // -- this is only to be really sure that we don't leave a client hanging
+        // forever.
+        LOG.warn("Unexpected error reading responses on connection " + this, t);
+        markClosed(new IOException("Error reading responses", t));
       }
       
       close();

src/test/core/org/apache/hadoop/ipc/TestIPC.java

@@ -249,6 +249,23 @@ public class TestIPC extends TestCase {
       throw new IOException(ERR_MSG);
     }
   }
+  
+  private static class LongRTEWritable extends LongWritable {
+    private final static String ERR_MSG = 
+      "Come across an runtime exception while reading";
+    
+    LongRTEWritable() {}
+    
+    LongRTEWritable(long longValue) {
+      super(longValue);
+    }
+    
+    public void readFields(DataInput in) throws IOException {
+      super.readFields(in);
+      throw new RuntimeException(ERR_MSG);
+    }
+  }
+
   public void testErrorClient() throws Exception {
     // start server
     Server server = new TestServer(1, false);
@@ -268,6 +285,30 @@ public class TestIPC extends TestCase {
       assertEquals(LongErrorWritable.ERR_MSG, cause.getMessage());
     }
   }
+  
+  public void testRuntimeExceptionWritable() throws Exception {
+    // start server
+    Server server = new TestServer(1, false);
+    InetSocketAddress addr = NetUtils.getConnectAddress(server);
+    server.start();
+
+    // start client
+    Client client = new Client(LongRTEWritable.class, conf);
+    try {
+      client.call(new LongRTEWritable(RANDOM.nextLong()),
+              addr, null, null);
+      fail("Expected an exception to have been thrown");
+    } catch (IOException e) {
+      // check error
+      Throwable cause = e.getCause();
+      assertTrue(cause instanceof IOException);
+      // it's double-wrapped
+      Throwable cause2 = cause.getCause();
+      assertTrue(cause2 instanceof RuntimeException);
+
+      assertEquals(LongRTEWritable.ERR_MSG, cause2.getMessage());
+    }
+  }
 
   /**
    * Test that, if the socket factory throws an IOE, it properly propagates