Repository: skiptests/astraea (java)

Issue:
[SERIALIZATION] serialize/deserialize 模式
https://github.com/skiptests/astraea/pull/1647#discussion_r1162876653

現在物件轉換成 byte array 的方法都放在 `ByteUtils` 內，各個需要序列化的元件再使用該些方法來實作。
有需要序列化的元件(如：`common/backup/RecordWriterBuilder` 中的 record serialization) 也應該套用相同模式。

Pull request description:
[SERIALIZATION] use ByteUtils to replace kafka seralizer/deserializer
fix #1653

Patch:
diff --git a/common/src/main/java/org/astraea/common/ByteUtils.java b/common/src/main/java/org/astraea/common/ByteUtils.java
index 5c3b2419..1f92771d 100644
--- a/common/src/main/java/org/astraea/common/ByteUtils.java
+++ b/common/src/main/java/org/astraea/common/ByteUtils.java
@@ -38,18 +38,44 @@ import org.astraea.common.metrics.BeanObject;
 
 public final class ByteUtils {
 
-  // ----------------------------------Java Primitive------------------------------------------- //
+  // ----------------------------------[Java Primitive]----------------------------------//
 
   public static byte[] toBytes(short value) {
     return new byte[] {(byte) (value >>> 8), (byte) value};
   }
 
+  public static short toShort(byte[] value) {
+    if (value.length != Short.BYTES) {
+      throw new IllegalArgumentException(
+          "expected size: " + Short.BYTES + " but actual: " + value.length);
+    }
+    short r = 0;
+    for (byte b : value) {
+      r <<= 8;
+      r |= b & 0xFF;
+    }
+    return r;
+  }
+
   public static byte[] toBytes(int value) {
     return new byte[] {
       (byte) (value >>> 24), (byte) (value >>> 16), (byte) (value >>> 8), (byte) value
     };
   }
 
+  public static int toInteger(byte[] value) {
+    if (value.length != Integer.BYTES) {
+      throw new IllegalArgumentException(
+          "expected size: " + Integer.BYTES + " but actual: " + value.length);
+    }
+    int r = 0;
+    for (byte b : value) {
+      r <<= 8;
+      r |= b & 0xFF;
+    }
+    return r;
+  }
+
   public static byte[] toBytes(long value) {
     return new byte[] {
       (byte) (value >>> 56),
@@ -63,12 +89,25 @@ public final class ByteUtils {
     };
   }
 
+  public static long toLong(byte[] value) {
+    if (value.length != Long.BYTES) {
+      throw new IllegalArgumentException(
+          "expected size: " + Long.BYTES + " but actual: " + value.length);
+    }
+    long r = 0;
+    for (byte b : value) {
+      r <<= 8;
+      r |= b & 0xFF;
+    }
+    return r;
+  }
+
   public static byte[] toBytes(String value) {
     return value.getBytes(StandardCharsets.UTF_8);
   }
 
-  public static byte[] toBytes(char value) {
-    return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
+  public static String toString(byte[] value) {
+    return new String(value, StandardCharsets.UTF_8);
   }
 
   public static byte[] toBytes(float value) {
@@ -78,6 +117,19 @@ public final class ByteUtils {
     };
   }
 
+  public static float toFloat(byte[] value) {
+    if (value.length != Float.BYTES) {
+      throw new IllegalArgumentException(
+          "expected size: " + Float.BYTES + " but actual: " + value.length);
+    }
+    int r = 0;
+    for (byte b : value) {
+      r <<= 8;
+      r |= b & 0xFF;
+    }
+    return Float.intBitsToFloat(r);
+  }
+
   public static byte[] toBytes(double value) {
     long longBits = Double.doubleToLongBits(value);
     return new byte[] {
@@ -92,6 +144,19 @@ public final class ByteUtils {
     };
   }
 
+  public static double toDouble(byte[] value) {
+    if (value.length != Double.BYTES) {
+      throw new IllegalArgumentException(
+          "expected size: " + Double.BYTES + " but actual: " + value.length);
+    }
+    long r = 0;
+    for (byte b : value) {
+      r <<= 8;
+      r |= b & 0xFF;
+    }
+    return Double.longBitsToDouble(r);
+  }
+
   public static byte[] toBytes(boolean value) {
     if (value) return new byte[] {1};
     return new byte[] {0};
diff --git a/common/src/main/java/org/astraea/common/consumer/Deserializer.java b/common/src/main/java/org/astraea/common/consumer/Deserializer.java
index 13e07537..f8021a19 100644
--- a/common/src/main/java/org/astraea/common/consumer/Deserializer.java
+++ b/common/src/main/java/org/astraea/common/consumer/Deserializer.java
@@ -19,12 +19,6 @@ package org.astraea.common.consumer;
 import java.util.Base64;
 import java.util.List;
 import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.DoubleDeserializer;
-import org.apache.kafka.common.serialization.FloatDeserializer;
-import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.apache.kafka.common.serialization.LongDeserializer;
-import org.apache.kafka.common.serialization.StringDeserializer;
 import org.astraea.common.ByteUtils;
 import org.astraea.common.Header;
 import org.astraea.common.json.JsonConverter;
@@ -61,20 +55,18 @@ public interface Deserializer<T> {
     };
   }
 
-  private static <T> Deserializer<T> of(
-      org.apache.kafka.common.serialization.Deserializer<T> deserializer) {
-    // the headers are not used by primitive type deserializer
-    return (topic, headers, data) -> deserializer.deserialize(topic, data);
-  }
-
   Deserializer<String> BASE64 =
       (topic, headers, data) -> data == null ? null : Base64.getEncoder().encodeToString(data);
-  Deserializer<byte[]> BYTE_ARRAY = of(new ByteArrayDeserializer());
-  Deserializer<String> STRING = of(new StringDeserializer());
-  Deserializer<Integer> INTEGER = of(new IntegerDeserializer());
-  Deserializer<Long> LONG = of(new LongDeserializer());
-  Deserializer<Float> FLOAT = of(new FloatDeserializer());
-  Deserializer<Double> DOUBLE = of(new DoubleDeserializer());
+  Deserializer<byte[]> BYTE_ARRAY = (topic, headers, data) -> data;
+  Deserializer<String> STRING =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toString(data);
+  Deserializer<Integer> INTEGER =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toInteger(data);
+  Deserializer<Long> LONG = (topic, headers, data) -> data == null ? null : ByteUtils.toLong(data);
+  Deserializer<Float> FLOAT =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toFloat(data);
+  Deserializer<Double> DOUBLE =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toDouble(data);
   Deserializer<BeanObject> BEAN_OBJECT = (topic, headers, data) -> ByteUtils.readBeanObject(data);
 
   /**
diff --git a/common/src/main/java/org/astraea/common/producer/Serializer.java b/common/src/main/java/org/astraea/common/producer/Serializer.java
index d3095f31..14126ff6 100644
--- a/common/src/main/java/org/astraea/common/producer/Serializer.java
+++ b/common/src/main/java/org/astraea/common/producer/Serializer.java
@@ -20,12 +20,6 @@ import java.nio.charset.StandardCharsets;
 import java.util.Collection;
 import java.util.List;
 import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.serialization.DoubleSerializer;
-import org.apache.kafka.common.serialization.FloatSerializer;
-import org.apache.kafka.common.serialization.IntegerSerializer;
-import org.apache.kafka.common.serialization.LongSerializer;
-import org.apache.kafka.common.serialization.StringSerializer;
 import org.astraea.common.ByteUtils;
 import org.astraea.common.Header;
 import org.astraea.common.Utils;
@@ -61,18 +55,15 @@ public interface Serializer<T> {
     };
   }
 
-  private static <T> Serializer<T> of(
-      org.apache.kafka.common.serialization.Serializer<T> serializer) {
-    // the headers are not used by primitive type serializer
-    return (topic, headers, data) -> serializer.serialize(topic, data);
-  }
-
-  Serializer<byte[]> BYTE_ARRAY = of(new ByteArraySerializer());
-  Serializer<String> STRING = of(new StringSerializer());
-  Serializer<Integer> INTEGER = of(new IntegerSerializer());
-  Serializer<Long> LONG = of(new LongSerializer());
-  Serializer<Float> FLOAT = of(new FloatSerializer());
-  Serializer<Double> DOUBLE = of(new DoubleSerializer());
+  Serializer<byte[]> BYTE_ARRAY = (topic, headers, data) -> data;
+  Serializer<String> STRING =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toBytes(data);
+  Serializer<Integer> INTEGER =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toBytes(data);
+  Serializer<Long> LONG = (topic, headers, data) -> data == null ? null : ByteUtils.toBytes(data);
+  Serializer<Float> FLOAT = (topic, headers, data) -> data == null ? null : ByteUtils.toBytes(data);
+  Serializer<Double> DOUBLE =
+      (topic, headers, data) -> data == null ? null : ByteUtils.toBytes(data);
   Serializer<BeanObject> BEAN_OBJECT = (topic, headers, data) -> ByteUtils.toBytes(data);
 
   /**


Test patch:
diff --git a/common/src/test/java/org/astraea/common/ByteUtilsTest.java b/common/src/test/java/org/astraea/common/ByteUtilsTest.java
index 73388e41..7cdef95b 100644
--- a/common/src/test/java/org/astraea/common/ByteUtilsTest.java
+++ b/common/src/test/java/org/astraea/common/ByteUtilsTest.java
@@ -26,37 +26,46 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class ByteUtilsTest {
+  @Test
+  void testShort2Bytes() {
+    Assertions.assertEquals(379, ByteUtils.toShort(ByteUtils.toBytes((short) 379)));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> ByteUtils.toShort(new byte[0]));
+  }
+
   @Test
   void testInt2Bytes() {
     Assertions.assertArrayEquals(new byte[] {0, 0, 4, -46}, ByteUtils.toBytes(1234));
+    Assertions.assertEquals(379, ByteUtils.toInteger(ByteUtils.toBytes(379)));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> ByteUtils.toInteger(new byte[0]));
   }
 
   @Test
   void testString2Bytes() {
     var string = Utils.randomString();
     Assertions.assertEquals(string, new String(ByteUtils.toBytes(string), StandardCharsets.UTF_8));
-  }
-
-  @Test
-  void testChar2Bytes() {
-    Assertions.assertEquals(
-        'c', new String(ByteUtils.toBytes('c'), StandardCharsets.UTF_8).charAt(0));
+    Assertions.assertEquals("xxx", ByteUtils.toString(ByteUtils.toBytes("xxx")));
   }
 
   @Test
   void testLong2Bytes() {
     Assertions.assertArrayEquals(new byte[] {0, 0, 0, 0, 0, 0, 1, 123}, ByteUtils.toBytes(379L));
+    Assertions.assertEquals(379L, ByteUtils.toLong(ByteUtils.toBytes(379L)));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> ByteUtils.toLong(new byte[0]));
   }
 
   @Test
   void testFloat2Bytes() {
     Assertions.assertArrayEquals(new byte[] {63, -64, 0, 0}, ByteUtils.toBytes(1.5f));
+    Assertions.assertEquals(379F, ByteUtils.toFloat(ByteUtils.toBytes(379F)));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> ByteUtils.toFloat(new byte[0]));
   }
 
   @Test
   void testDouble2Bytes() {
     Assertions.assertArrayEquals(
         new byte[] {64, 94, -58, 102, 102, 102, 102, 102}, ByteUtils.toBytes(123.1D));
+    Assertions.assertEquals(379D, ByteUtils.toDouble(ByteUtils.toBytes(379D)));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> ByteUtils.toDouble(new byte[0]));
   }
 
   @Test