[Improve] update avro bytes type convert

liunaijie · Jan 5, 2024 · 7235aed · 7235aed
1 parent b891202
commit 7235aed
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 5 deletions.
diff --git a/...e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java b/...e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -68,7 +68,6 @@
 
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -385,7 +384,24 @@ public void testFakeSourceToKafkaAvroFormat(TestContainer container)
                         });
         Assertions.assertEquals(90, kafkaSTRow.size());
         kafkaSTRow.forEach(
-                row -> Assertions.assertEquals("fake_source_avro", row.getField(2).toString()));
+                row -> {
+                    Assertions.assertInstanceOf(Map.class, row.getField(0));
+                    Assertions.assertInstanceOf(Integer[].class, row.getField(1));
+                    Assertions.assertInstanceOf(String.class, row.getField(2));
+                    Assertions.assertEquals("fake_source_avro", row.getField(2).toString());
+                    Assertions.assertInstanceOf(Boolean.class, row.getField(3));
+                    Assertions.assertInstanceOf(Byte.class, row.getField(4));
+                    Assertions.assertInstanceOf(Short.class, row.getField(5));
+                    Assertions.assertInstanceOf(Integer.class, row.getField(6));
+                    Assertions.assertInstanceOf(Long.class, row.getField(7));
+                    Assertions.assertInstanceOf(Float.class, row.getField(8));
+                    Assertions.assertInstanceOf(Double.class, row.getField(9));
+                    Assertions.assertInstanceOf(byte[].class, row.getField(10));
+                    Assertions.assertInstanceOf(LocalDate.class, row.getField(11));
+                    Assertions.assertInstanceOf(BigDecimal.class, row.getField(12));
+                    Assertions.assertInstanceOf(LocalDateTime.class, row.getField(13));
+                    Assertions.assertInstanceOf(SeaTunnelRow.class, row.getField(14));
+                });
     }
 
     @TestTemplate
@@ -435,8 +451,7 @@ public void testKafkaAvroToAssert(TestContainer container)
                     Assertions.assertEquals(Float.parseFloat("1.1"), row.getField(9));
                     Assertions.assertEquals(Double.parseDouble("1.1"), row.getField(10));
                     Assertions.assertEquals(BigDecimal.valueOf(11, 1), row.getField(11));
-                    Assertions.assertArrayEquals(
-                            "test".getBytes(), ((ByteBuffer) row.getField(12)).array());
+                    Assertions.assertArrayEquals("test".getBytes(), (byte[]) row.getField(12));
                     Assertions.assertEquals(LocalDate.of(2024, 1, 1), row.getField(13));
                     Assertions.assertEquals(
                             LocalDateTime.of(2024, 1, 1, 12, 59, 23), row.getField(14));

diff --git a/...tunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java b/...tunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
@@ -35,6 +35,7 @@
 
 import java.io.Serializable;
 import java.lang.reflect.Array;
+import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -91,11 +92,12 @@ private Object convertField(SeaTunnelDataType<?> dataType, Object val) {
             case FLOAT:
             case DOUBLE:
             case NULL:
-            case BYTES:
             case DATE:
             case DECIMAL:
             case TIMESTAMP:
                 return val;
+            case BYTES:
+                return ((ByteBuffer) val).array();
             case SMALLINT:
                 return ((Integer) val).shortValue();
             case TINYINT: