[HUDI-1217] Improve avroToBytes method of HoodieAvroUtils (#2018)

hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java

@@ -83,15 +83,17 @@ public class HoodieAvroUtils {
   /**
    * Convert a given avro record to bytes.
    */
-  public static byte[] avroToBytes(GenericRecord record) throws IOException {
+  public static byte[] avroToBytes(GenericRecord record) {
     GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<>(record.getSchema());
-    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
       BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, reuseEncoder.get());
       reuseEncoder.set(encoder);
       writer.write(record, encoder);
       encoder.flush();
-    out.close();
       return out.toByteArray();
+    } catch (IOException e) {
+      throw new HoodieIOException("Cannot convert GenericRecord to bytes", e);
+    }
   }
 
   /**

hudi-common/src/main/java/org/apache/hudi/common/model/BaseAvroPayload.java

@@ -20,11 +20,9 @@ package org.apache.hudi.common.model;
 
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.exception.HoodieIOException;
 
 import org.apache.avro.generic.GenericRecord;
 
-import java.io.IOException;
 import java.io.Serializable;
 
 /**
@@ -48,11 +46,7 @@ public abstract class BaseAvroPayload implements Serializable {
    * @param orderingVal {@link Comparable} to be used in pre combine.
    */
   public BaseAvroPayload(GenericRecord record, Comparable orderingVal) {
-    try {
     this.recordBytes = record != null ? HoodieAvroUtils.avroToBytes(record) : new byte[0];
-    } catch (IOException io) {
-      throw new HoodieIOException("Cannot convert GenericRecord to bytes", io);
-    }
     this.orderingVal = orderingVal;
     if (orderingVal == null) {
       throw new HoodieException("Ordering value is null for record: " + record);

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroPayload.java

@@ -20,7 +20,6 @@ package org.apache.hudi.common.model;
 
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.common.util.Option;
-import org.apache.hudi.exception.HoodieIOException;
 
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericRecord;
@@ -39,15 +38,11 @@ public class HoodieAvroPayload implements HoodieRecordPayload<HoodieAvroPayload>
   private final byte[] recordBytes;
 
   public HoodieAvroPayload(Option<GenericRecord> record) {
-    try {
     if (record.isPresent()) {
       this.recordBytes = HoodieAvroUtils.avroToBytes(record.get());
     } else {
       this.recordBytes = new byte[0];
     }
-    } catch (IOException io) {
-      throw new HoodieIOException("Cannot convert record to bytes", io);
-    }
   }
 
   @Override

hudi-common/src/test/java/org/apache/hudi/common/testutils/AvroBinaryTestPayload.java

@@ -21,7 +21,6 @@ package org.apache.hudi.common.testutils;
 import org.apache.hudi.avro.HoodieAvroUtils;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.util.Option;
-import org.apache.hudi.exception.HoodieIOException;
 
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericRecord;
@@ -37,16 +36,11 @@ public class AvroBinaryTestPayload implements HoodieRecordPayload {
   private final byte[] recordBytes;
 
   public AvroBinaryTestPayload(Option<GenericRecord> record) {
-
-    try {
     if (record.isPresent()) {
       recordBytes = HoodieAvroUtils.avroToBytes(record.get());
     } else {
       recordBytes = new byte[0];
     }
-    } catch (IOException io) {
-      throw new HoodieIOException("unable to convert payload to bytes");
-    }
   }
 
   @Override