Repository: grpc/grpc-java
Issue: NPE in ProtoInputStream.drainTo

```
java.lang.NullPointerException: null
        at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:210)
        at com.google.common.io.ByteStreams.copy(ByteStreams.java:65)
        at io.grpc.protobuf.ProtoInputStream.drainTo(ProtoInputStream.java:74)
        at io.grpc.internal.MessageFramer.writeToOutputStream(MessageFramer.java:224)
        at io.grpc.internal.MessageFramer.writeKnownLength(MessageFramer.java:185)
        at io.grpc.internal.MessageFramer.writeUncompressed(MessageFramer.java:135)
        at io.grpc.internal.MessageFramer.writePayload(MessageFramer.java:125)
        at io.grpc.internal.AbstractStream.writeMessage(AbstractStream.java:165)
        at io.grpc.internal.ClientCallImpl.sendMessage(ClientCallImpl.java:204)
        at com.google.cloud.bigtable.grpc.io.ReconnectingChannel$DelayingCall.sendMessage(ReconnectingChannel.java:122)
        at io.grpc.ForwardingClientCall.sendMessage(ForwardingClientCall.java:65)
```

It looks like if you call `drainTo()` twice then it will NPE. Calling `drainTo()` twice is sort of strange, so I wouldn't expect this to be critical, but we should still not NPE.


Changed files:
protobuf/src/main/java/io/grpc/protobuf/ProtoInputStream.java, protobuf/src/test/java/io/grpc/protobuf/ProtoUtilsTest.java

diff --git a/protobuf/src/main/java/io/grpc/protobuf/ProtoInputStream.java b/protobuf/src/main/java/io/grpc/protobuf/ProtoInputStream.java
index d5ebae188..186e3c12e 100644
--- a/protobuf/src/main/java/io/grpc/protobuf/ProtoInputStream.java
+++ b/protobuf/src/main/java/io/grpc/protobuf/ProtoInputStream.java
@@ -70,9 +70,11 @@ class ProtoInputStream extends InputStream implements Drainable, KnownLength {
       written = message.getSerializedSize();
       message.writeTo(target);
       message = null;
-    } else {
+    } else if (partial != null) {
       written = (int) ByteStreams.copy(partial, target);
       partial = null;
+    } else {
+      written = 0;
     }
     return written;
   }
diff --git a/protobuf/src/test/java/io/grpc/protobuf/ProtoUtilsTest.java b/protobuf/src/test/java/io/grpc/protobuf/ProtoUtilsTest.java
index 7fc93e023..7858568c5 100644
--- a/protobuf/src/test/java/io/grpc/protobuf/ProtoUtilsTest.java
+++ b/protobuf/src/test/java/io/grpc/protobuf/ProtoUtilsTest.java
@@ -41,6 +41,7 @@ import com.google.protobuf.Empty;
 import com.google.protobuf.Enum;
 import com.google.protobuf.Type;
 
+import io.grpc.Drainable;
 import io.grpc.MethodDescriptor.Marshaller;
 
 import org.junit.Test;
@@ -48,6 +49,7 @@ import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
@@ -115,4 +117,46 @@ public class ProtoUtilsTest {
     assertEquals(-1, is.read());
     assertEquals(0, is.available());
   }
+
+  @Test
+  public void testDrainTo_all() throws Exception {
+    byte[] golden = ByteStreams.toByteArray(marshaller.stream(proto));
+    InputStream is = marshaller.stream(proto);
+    Drainable d = (Drainable) is;
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    int drained = d.drainTo(baos);
+    assertEquals(baos.size(), drained);
+    assertArrayEquals(golden, baos.toByteArray());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void testDrainTo_partial() throws Exception {
+    final byte[] golden;
+    {
+      InputStream is = marshaller.stream(proto);
+      is.read();
+      golden = ByteStreams.toByteArray(is);
+    }
+    InputStream is = marshaller.stream(proto);
+    is.read();
+    Drainable d = (Drainable) is;
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    int drained = d.drainTo(baos);
+    assertEquals(baos.size(), drained);
+    assertArrayEquals(golden, baos.toByteArray());
+    assertEquals(0, is.available());
+  }
+
+  @Test
+  public void testDrainTo_none() throws Exception {
+    byte[] golden = ByteStreams.toByteArray(marshaller.stream(proto));
+    InputStream is = marshaller.stream(proto);
+    ByteStreams.toByteArray(is);
+    Drainable d = (Drainable) is;
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    assertEquals(0, d.drainTo(baos));
+    assertArrayEquals(new byte[0], baos.toByteArray());
+    assertEquals(0, is.available());
+  }
 }