Comments and more tests

lang/java/avro/pom.xml

@@ -250,5 +250,10 @@
       <artifactId>hamcrest-library</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

lang/java/avro/src/main/java/org/apache/avro/io/BinaryEncoder.java

@@ -48,7 +48,7 @@ public void writeString(Utf8 utf8) throws IOException {
 
   @Override
   public void writeString(String string) throws IOException {
-    if (string.isEmpty()) {
+    if (0 == string.length()) {
       writeZero();
       return;
     }

lang/java/avro/src/main/java/org/apache/avro/io/BlockingDirectBinaryEncoder.java

@@ -25,23 +25,28 @@
 /**
  * An {@link Encoder} for Avro's binary encoding that does not buffer output.
  * <p/>
- * This encoder does not buffer writes, and as a result is slower than
- * {@link BufferedBinaryEncoder}. However, it is lighter-weight and useful when
- * the buffering in BufferedBinaryEncoder is not desired and/or the Encoder is
- * very short-lived.
+ * This encoder does not buffer writes in contrast to
+ * {@link BufferedBinaryEncoder}. However, it is lighter-weight and useful when:
+ * The buffering in BufferedBinaryEncoder is not desired because you buffer a
+ * different level or the Encoder is very short-lived.
+ * </p>
+ * The BlockingDirectBinaryEncoder will encode the number of bytes of the Map
+ * and Array blocks. This will allow to postpone the decoding, or skip over it
+ * at all.
  * <p/>
  * To construct, use
  * {@link EncoderFactory#blockingDirectBinaryEncoder(OutputStream, BinaryEncoder)}
  * <p/>
- * BlockingDirectBinaryEncoder is not thread-safe
+ * {@link BlockingDirectBinaryEncoder} instances returned by this method are not
+ * thread-safe
  *
  * @see BinaryEncoder
  * @see EncoderFactory
  * @see Encoder
  * @see Decoder
  */
 public class BlockingDirectBinaryEncoder extends DirectBinaryEncoder {
-  private static final ThreadLocal<BufferOutputStream> BUFFER = ThreadLocal.withInitial(BufferOutputStream::new);
+  private final BufferOutputStream buffer;
 
   private OutputStream originalStream;
 
@@ -57,37 +62,36 @@ public class BlockingDirectBinaryEncoder extends DirectBinaryEncoder {
    */
   public BlockingDirectBinaryEncoder(OutputStream out) {
     super(out);
+    this.buffer = new BufferOutputStream();
   }
 
   private void startBlock() {
     if (inBlock) {
       throw new RuntimeException("Nested Maps/Arrays are not supported by the BlockingDirectBinaryEncoder");
     }
     originalStream = out;
-    BufferOutputStream buf = BUFFER.get();
-    buf.reset();
-    out = buf;
+    buffer.reset();
+    out = buffer;
     inBlock = true;
   }
 
   private void endBlock() {
     if (!inBlock) {
       throw new RuntimeException("Called endBlock, while not buffering a block");
     }
-    BufferOutputStream buf = (BufferOutputStream) out;
     out = originalStream;
     if (blockItemCount > 0) {
       try {
         // Make it negative, so the reader knows that the number of bytes is coming
         writeLong(-blockItemCount);
-        writeLong(buf.size());
-        writeFixed(buf.toBufferWithoutCopy());
+        writeLong(buffer.size());
+        writeFixed(buffer.toBufferWithoutCopy());
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
     }
     inBlock = false;
-    buf.reset();
+    buffer.reset();
   }
 
   @Override

lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryEncoderFidelity.java

@@ -221,8 +221,6 @@ void blockingDirectBinaryEncoder() throws IOException {
     // 0: 0 elements in the block
     assertArrayEquals(baos.toByteArray(), new byte[] { 0 });
     baos.reset();
-
-    baos.reset();
   }
 
   @Test

lang/java/avro/src/test/java/org/apache/avro/io/TestBlockingDirectBinaryEncoder.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.avro.io;
+
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaNormalization;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.specific.TestRecordWithMapsAndArrays;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.mockito.Mockito.*;
+
+public class TestBlockingDirectBinaryEncoder {
+
+  @Test
+  void blockingDirectBinaryEncoder() throws IOException, NoSuchAlgorithmException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    BinaryEncoder encoder = EncoderFactory.get().blockingDirectBinaryEncoder(baos, null);
+
+    // This is needed because there is no BlockingDirectBinaryEncoder
+    // BinaryMessageWriter
+    // available out of the box
+    encoder.writeFixed(new byte[] { (byte) 0xC3, (byte) 0x01 });
+    encoder.writeFixed(SchemaNormalization.parsingFingerprint("CRC-64-AVRO", TestRecordWithMapsAndArrays.SCHEMA$));
+
+    int len = 5;
+
+    encoder.writeArrayStart();
+    encoder.setItemCount(len);
+    for (int i = 0; i < len; i++) {
+      encoder.startItem();
+      encoder.writeString(Integer.toString(i));
+    }
+    encoder.writeArrayEnd();
+
+    encoder.writeMapStart();
+    encoder.setItemCount(len);
+    for (long i = 0; i < len; i++) {
+      encoder.startItem();
+      encoder.writeString(Long.toString(i));
+      encoder.writeLong(i);
+    }
+    encoder.writeMapEnd();
+    encoder.flush();
+
+    BinaryMessageDecoder<TestRecordWithMapsAndArrays> decoder = TestRecordWithMapsAndArrays.getDecoder();
+    TestRecordWithMapsAndArrays r = decoder.decode(baos.toByteArray());
+
+    assertThat(r.getArr(), is(Arrays.asList("0", "1", "2", "3", "4")));
+    Map<String, Long> map = r.getMap();
+    assertThat(map.size(), is(5));
+    for (long i = 0; i < len; i++) {
+      assertThat(map.get(Long.toString(i)), is(i));
+    }
+  }
+
+  @Test
+  void testSkippingUsingBlocks() throws IOException, NoSuchAlgorithmException {
+    // Create an empty schema for read, so we skip over all the fields
+    Schema emptySchema = new Schema.Parser().parse(
+        "{\"type\":\"record\",\"name\":\"TestRecordWithMapsAndArrays\",\"namespace\":\"org.apache.avro.specific\",\"fields\":[]}");
+
+    GenericDatumReader<?> in = new GenericDatumReader<>(TestRecordWithMapsAndArrays.SCHEMA$, emptySchema);
+    Decoder mockDecoder = mock(BinaryDecoder.class);
+
+    for (long i = 0; i < 1; i++) {
+      in.read(null, mockDecoder);
+    }
+
+    verify(mockDecoder, times(1)).skipMap();
+    verify(mockDecoder, times(1)).skipArray();
+    verify(mockDecoder, times(0)).readString();
+    verify(mockDecoder, times(0)).readLong();
+  }
+}