AVRO-4073: Create Convenience toBytes Method for Datum Writer (#3197)

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

@@ -17,6 +17,7 @@
  */
 package org.apache.avro.io;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 
 import org.apache.avro.Schema;
@@ -36,4 +37,19 @@ public interface DatumWriter<D> {
    * the schema from the datum to the output.
    */
   void write(D datum, Encoder out) throws IOException;
+
+  /**
+   * Convenience method to Write a datum to a byte array. Traverse the schema,
+   * depth first, writing each leaf value in the schema from the datum to the byte
+   * array.
+   *
+   * @param datum The datum to serialize
+   * @return The serialized datum stored in an array of bytes
+   */
+  default byte[] toByteArray(D datum) throws IOException {
+    try (ByteArrayOutputStream out = new ByteArrayOutputStream(128)) {
+      write(datum, EncoderFactory.get().directBinaryEncoder(out, null));
+      return out.toByteArray();
+    }
+  }
 }

lang/java/avro/src/test/java/org/apache/avro/specific/TestSpecificData.java

@@ -179,6 +179,20 @@ void nonStringable() throws Exception {
     }
   }
 
+  @Test
+  void testToByteArray() throws Exception {
+    final Schema string = Schema.create(Type.STRING);
+    final DatumWriter<String> writer = new SpecificDatumWriter<>(string);
+
+    try (final ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+      final Encoder encoder = EncoderFactory.get().directBinaryEncoder(baos, null);
+      writer.write("test", encoder);
+
+      final byte[] bytes = writer.toByteArray("test");
+      assertArrayEquals(baos.toByteArray(), bytes);
+    }
+  }
+
   @Test
   void classNameContainingReservedWords() {
     final Schema schema = Schema.createRecord("AnyName", null, "db.public.table", false);