feat: maps AVRO BYTES ByteBuffer column (#1893)

* feat: maps AVRO BYTES ByteBuffer column

AVRO GenericData uses java.nio.ByteBuffer as a representation of BYTES
data (see
https://github.com/apache/avro/blob/main/lang/java/avro/src/main/java/org/apache/avro/generic/GenericData.java#L1093).

This PR maps a ByteBuffer record into a byte array so it can be written
into Spanner.

* fix: default to bytebuffer instead of byte[]

* fix: update error message

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/avro/AvroToValueMapper.java

@@ -24,6 +24,7 @@
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.math.RoundingMode;
+import java.nio.ByteBuffer;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
@@ -281,7 +282,7 @@ static ByteArray avroFieldToByteArray(Object recordValue, Schema fieldSchema) {
         }
         return ByteArray.copyFrom(Hex.decodeHex(s));
       }
-      return ByteArray.copyFrom((byte[]) recordValue);
+      return ByteArray.copyFrom(((ByteBuffer) recordValue).array());
     } catch (Exception e) {
       throw new AvroTypeConvertorException(
           "Unable to convert "

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/AvroToValueMapperTest.java

@@ -24,6 +24,7 @@
 import com.google.cloud.teleport.v2.spanner.migrations.exceptions.AvroTypeConvertorException;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
+import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.avro.Schema;
@@ -311,10 +312,11 @@ public void testAvroFieldToByteArray_StringInput() throws Exception {
 
   @Test
   public void testAvroFieldToByteArray_ValidByteArrayInput() {
-    byte[] inputValue = {10, 20, 30};
+    byte[] bytes = {10, 20, 30};
+    ByteBuffer inputValue = ByteBuffer.wrap(bytes);
     ByteArray result =
         AvroToValueMapper.avroFieldToByteArray(inputValue, SchemaBuilder.builder().bytesType());
-    assertEquals(ByteArray.copyFrom(inputValue), result);
+    assertEquals(ByteArray.copyFrom(bytes), result);
   }
 
   @Test

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java

@@ -389,7 +389,7 @@ public void transformChangeEventTest_identityMapper() throws InvalidTransformati
     genericRecord.put("string_col", "hello");
     genericRecord.put(
         "numeric_col", ByteBuffer.wrap(new BigDecimal("12.34").unscaledValue().toByteArray()));
-    genericRecord.put("bytes_col", new byte[] {10, 20, 30});
+    genericRecord.put("bytes_col", ByteBuffer.wrap(new byte[] {10, 20, 30}));
     genericRecord.put("timestamp_col", 1602599400056483L);
     genericRecord.put("date_col", 738991);
     GenericRecordTypeConvertor genericRecordTypeConvertor =