core: Fix a bug for exception handling at messageRead

Found this bug in some unit test in which client-streaming call is hanging because `ServerCallImpl#messageRead()` did not handle RuntimeException properly

Author: dapengzhang0

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -245,12 +245,13 @@ public void messageRead(final InputStream message) {
         try {
           message.close();
         } catch (IOException e) {
+          throw new RuntimeException(e);
+        } finally {
           if (t != null) {
             // TODO(carl-mastrangelo): Maybe log e here.
             Throwables.propagateIfPossible(t);
             throw new RuntimeException(t);
           }
-          throw new RuntimeException(e);
         }
       }
     }

core/src/test/java/io/grpc/internal/ServerCallImplTest.java

@@ -37,6 +37,7 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.verify;
@@ -311,6 +312,20 @@ public void streamListener_messageRead_onlyOnce() {
     verify(callListener).onMessage(1234L);
   }
 
+  @Test
+  public void streamListener_unexpectedRuntimeException() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<Long>(
+            call, callListener, context, statsTraceCtx);
+    doThrow(new RuntimeException("unexpected exception"))
+        .when(callListener)
+        .onMessage(any(Long.class));
+
+    thrown.expect(RuntimeException.class);
+    thrown.expectMessage("unexpected exception");
+    streamListener.messageRead(method.streamRequest(1234L));
+  }
+
   private void checkStats(Status.Code statusCode) {
     CensusTestUtils.MetricsRecord record = censusCtxFactory.pollRecord();
     assertNotNull(record);