Optimise writing of serialised Work Unit to File system

Author: thisisArjit

gobblin-api/src/main/java/org/apache/gobblin/compat/hadoop/TextSerializer.java

@@ -31,7 +31,7 @@ public class TextSerializer {
    */
   public static void writeStringAsText(DataOutput stream, String str) throws IOException {
     writeVLong(stream, str.length());
-    stream.writeBytes(str);
+    stream.writeChars(str);
   }
 
   /**
@@ -42,7 +42,7 @@ public static String readTextAsString(DataInput in) throws IOException {
     StringBuilder sb = new StringBuilder();
 
     for (int i = 0; i < bufLen; i++) {
-      sb.append((char) in.readByte());
+      sb.append(in.readChar());
     }
     return sb.toString();
   }

gobblin-api/src/test/java/org/apache/gobblin/compat/TextSerializerTest.java

@@ -22,7 +22,6 @@
 import java.io.DataOutputStream;
 import java.io.IOException;
 
-import org.apache.hadoop.io.Text;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -35,9 +34,7 @@ public class TextSerializerTest {
   private static final String[] textsToSerialize = new String[]{"abracadabra", Strings.repeat("longString", 128000)};
 
   @Test
-  public void testSerialize()
-      throws IOException {
-
+  public void testWriteAndRead() throws IOException {
     // Use our serializer, verify Hadoop deserializer can read it back
     for (String textToSerialize : textsToSerialize) {
       ByteArrayOutputStream bOs = new ByteArrayOutputStream();
@@ -49,31 +46,8 @@ public void testSerialize()
       ByteArrayInputStream bIn = new ByteArrayInputStream(bOs.toByteArray());
       DataInputStream dataInputStream = new DataInputStream(bIn);
 
-      Text hadoopText = new Text();
-      hadoopText.readFields(dataInputStream);
-
-      Assert.assertEquals(hadoopText.toString(), textToSerialize);
-    }
-  }
-
-  @Test
-  public void testDeserialize() throws IOException {
-    // Use Hadoop's serializer, verify our deserializer can read the string back
-    for (String textToSerialize : textsToSerialize) {
-      ByteArrayOutputStream bOs = new ByteArrayOutputStream();
-      DataOutputStream dataOutputStream = new DataOutputStream(bOs);
-
-      Text hadoopText = new Text();
-      hadoopText.set(textToSerialize);
-      hadoopText.write(dataOutputStream);
-      dataOutputStream.close();
-
-      ByteArrayInputStream bIn = new ByteArrayInputStream(bOs.toByteArray());
-      DataInputStream dataInputStream = new DataInputStream(bIn);
-
-      String deserializedString = TextSerializer.readTextAsString(dataInputStream);
-
-      Assert.assertEquals(deserializedString, textToSerialize);
+      String deserializedText = TextSerializer.readTextAsString(dataInputStream);
+      Assert.assertEquals(deserializedText, textToSerialize);
     }
   }
 }