Write record batch size prefix in Java

Change-Id: I057c329090faa0c2eb5fb65f8566c17130dd2a69

java/vector/src/main/java/org/apache/arrow/vector/file/ArrowReader.java

@@ -123,7 +123,11 @@ public class ArrowReader implements AutoCloseable {
     if (n != l) {
       throw new IllegalStateException(n + " != " + l);
     }
-    RecordBatch recordBatchFB = RecordBatch.getRootAsRecordBatch(buffer.nioBuffer().asReadOnlyBuffer());
+
+    // Record batch flatbuffer is prefixed by its size as int32le
+    final ArrowBuf metadata = buffer.slice(4, recordBatchBlock.getMetadataLength() - 4);
+    RecordBatch recordBatchFB = RecordBatch.getRootAsRecordBatch(metadata.nioBuffer().asReadOnlyBuffer());
+
     int nodesLength = recordBatchFB.nodesLength();
     final ArrowBuf body = buffer.slice(recordBatchBlock.getMetadataLength(), (int)recordBatchBlock.getBodyLength());
     List<ArrowFieldNode> nodes = new ArrayList<>();

java/vector/src/main/java/org/apache/arrow/vector/file/ArrowWriter.java

@@ -99,9 +99,10 @@ public class ArrowWriter implements AutoCloseable {
   public void writeRecordBatch(ArrowRecordBatch recordBatch) throws IOException {
     checkStarted();
     align();
-    // write metadata header
+
+    // write metadata header with int32 size prefix
     long offset = currentPosition;
-    write(recordBatch);
+    write(recordBatch, true);
     align();
     // write body
     long bodyOffset = currentPosition;
@@ -166,14 +167,21 @@ public class ArrowWriter implements AutoCloseable {
 
   private void writeFooter() throws IOException {
     // TODO: dictionaries
-    write(new ArrowFooter(schema, Collections.<ArrowBlock>emptyList(), recordBatches));
+    write(new ArrowFooter(schema, Collections.<ArrowBlock>emptyList(), recordBatches), false);
   }
 
-  private long write(FBSerializable writer) throws IOException {
+  private long write(FBSerializable writer, boolean withSizePrefix) throws IOException {
     FlatBufferBuilder builder = new FlatBufferBuilder();
     int root = writer.writeTo(builder);
     builder.finish(root);
-    return write(builder.dataBuffer());
+
+    ByteBuffer buffer = builder.dataBuffer();
+
+    if (withSizePrefix) {
+      writeIntLittleEndian(buffer.remaining());
+    }
+
+    return write(buffer);
   }
 
 }