[beam] branch master updated: [BEAM-5439] fix performance issue in StringUtf8Coder (#6812)

[lc...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new fbf82c7  [BEAM-5439] fix performance issue in StringUtf8Coder (#6812)
fbf82c7 is described below

commit fbf82c793126cec0370adc63679b515820026b11
Author: Julien Tournay <bo...@gmail.com>
AuthorDate: Thu Oct 25 00:30:50 2018 +0200

    [BEAM-5439] fix performance issue in StringUtf8Coder (#6812)
---
 .../java/org/apache/beam/sdk/coders/StringUtf8Coder.java    | 13 ++++++-------
 1 file changed, 6 insertions(+), 7 deletions(-)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
index 339b8f1..5710781 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
@@ -18,8 +18,7 @@
 package org.apache.beam.sdk.coders;
 
 import com.google.common.base.Utf8;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
+import com.google.common.io.ByteStreams;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
@@ -46,19 +45,19 @@ public class StringUtf8Coder extends AtomicCoder<String> {
   private static final StringUtf8Coder INSTANCE = new StringUtf8Coder();
   private static final TypeDescriptor<String> TYPE_DESCRIPTOR = new TypeDescriptor<String>() {};
 
-  private static void writeString(String value, DataOutputStream dos) throws IOException {
+  private static void writeString(String value, OutputStream dos) throws IOException {
     byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
     VarInt.encode(bytes.length, dos);
     dos.write(bytes);
   }
 
-  private static String readString(DataInputStream dis) throws IOException {
+  private static String readString(InputStream dis) throws IOException {
     int len = VarInt.decodeInt(dis);
     if (len < 0) {
       throw new CoderException("Invalid encoded string length: " + len);
     }
     byte[] bytes = new byte[len];
-    dis.readFully(bytes);
+    ByteStreams.readFully(dis, bytes);
     return new String(bytes, StandardCharsets.UTF_8);
   }
 
@@ -82,7 +81,7 @@ public class StringUtf8Coder extends AtomicCoder<String> {
         outStream.write(bytes);
       }
     } else {
-      writeString(value, new DataOutputStream(outStream));
+      writeString(value, outStream);
     }
   }
 
@@ -98,7 +97,7 @@ public class StringUtf8Coder extends AtomicCoder<String> {
       return new String(bytes, StandardCharsets.UTF_8);
     } else {
       try {
-        return readString(new DataInputStream(inStream));
+        return readString(inStream);
       } catch (EOFException | UTFDataFormatException exn) {
         // These exceptions correspond to decoding problems, so change
         // what kind of exception they're branded as.