Empty stream double check.

Author: adampolomski

flight/flight-core/src/main/java/org/apache/arrow/flight/ArrowMessage.java

@@ -287,7 +287,11 @@ private static ArrowMessage frame(BufferAllocator allocator, final InputStream s
       ArrowBuf body = null;
       ArrowBuf appMetadata = null;
       while (stream.available() > 0) {
-        int tag = readRawVarint32(stream);
+        final int tagFirstByte = stream.read();
+        if (tagFirstByte == -1) {
+          break;
+        }
+        int tag = readRawVarint32(tagFirstByte, stream);
         switch (tag) {
           case DESCRIPTOR_TAG:
             {
@@ -366,6 +370,10 @@ private static ArrowMessage frame(BufferAllocator allocator, final InputStream s
 
   private static int readRawVarint32(InputStream is) throws IOException {
     int firstByte = is.read();
+    return readRawVarint32(firstByte, is);
+  }
+
+  private static int readRawVarint32(int firstByte, InputStream is) throws IOException {
     return CodedInputStream.readRawVarint32(firstByte, is);
   }