Core: Avro writers use BlockingBinaryEncoder to enable array/map sizes

When writing Avro files often Iceberg is writing arrays and maps. The current use of binaryEncoder() and directBinaryEncoder() of org.apache.avro.io.EncoderFactory do not write the length of the arrays or maps to Avro since the encoder does not buffer the output to calculate a length.

Knowing the length of an array or map is useful to clients decoding the Avro file since they can skip decoding the entire array or map if it is not needed when reading the file.  This PR changes all Avro writers to use blockingBinaryEncoder(), this encoder does not "block" in the concurrency sense but it does buffer the output of objects such that the lengths of arrays and maps will be calculated.

See:

https://avro.apache.org/docs/1.5.1/api/java/org/apache/avro/io/EncoderFactory.html#blockingBinaryEncoder(java.io.OutputStream,%20org.apache.avro.io.BinaryEncoder)

For details between the different Avro encoders.

core/src/main/java/org/apache/iceberg/avro/AvroEncoderUtil.java

@@ -54,7 +54,8 @@ public static <T> byte[] encode(T datum, Schema avroSchema) throws IOException {
       dataOut.writeUTF(avroSchema.toString());
 
       // Encode the datum with avro schema.
-      BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+      BinaryEncoder encoder =
+          EncoderFactory.configureBlockSize(1024 * 1024).blockingBinaryEncoder(out, null);
       DatumWriter<T> writer = new GenericAvroWriter<>(avroSchema);
       writer.write(datum, encoder);
       encoder.flush();

core/src/main/java/org/apache/iceberg/data/avro/IcebergEncoder.java

@@ -98,7 +98,8 @@ public ByteBuffer encode(D datum) throws IOException {
 
   @Override
   public void encode(D datum, OutputStream stream) throws IOException {
-    BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(stream, ENCODER.get());
+    BinaryEncoder encoder =
+        EncoderFactory.configureBlockSize(1024 * 1024).blockingBinaryEncoder(stream, ENCODER.get());
     ENCODER.set(encoder);
     writer.write(datum, encoder);
     encoder.flush();

core/src/main/java/org/apache/iceberg/encryption/KeyMetadataEncoder.java

@@ -90,7 +90,8 @@ public ByteBuffer encode(KeyMetadata datum) throws IOException {
 
   @Override
   public void encode(KeyMetadata datum, OutputStream stream) throws IOException {
-    BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(stream, ENCODER.get());
+    BinaryEncoder encoder =
+        EncoderFactory.configureBlockSize(1024 * 1024).blockingBinaryEncoder(stream, ENCODER.get());
     ENCODER.set(encoder);
     writer.write(datum, encoder);
     encoder.flush();