commit 5b81017a4d11c09cfa0180e3d56e22c22744559b
author Rajesh Balamohan <rbalamohan@apache.org> Wed Jan 22 03:55:46 2020 +0530
committer Rajesh Balamohan <rbalamohan@apache.org> Wed Jan 22 03:55:46 2020 +0530
tree c59eb017f67fb7a70d60f9b4f7f23611f3d3a3be
parent bf4942b49a3d9c0d326f64715b3e76952b77382d

TEZ-4113: TezUtils::createByteStringFromConf should use snappy instead of DeflaterOutputStream (rbalamohan, reviewed by Ashutosh Chauhan)

tez-api/src/main/java/org/apache/tez/common/TezUtils.java

diff --git a/tez-api/src/main/java/org/apache/tez/common/TezUtils.java b/tez-api/src/main/java/org/apache/tez/common/TezUtils.java
index 072c02f..192d2a0 100644
--- a/tez-api/src/main/java/org/apache/tez/common/TezUtils.java
+++ b/tez-api/src/main/java/org/apache/tez/common/TezUtils.java
@@ -25,9 +25,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.zip.Deflater;
-import java.util.zip.DeflaterOutputStream;
-import java.util.zip.InflaterInputStream;
 
 import com.google.common.base.Preconditions;
 import com.google.protobuf.ByteString;
@@ -42,6 +39,8 @@
 import org.apache.tez.dag.api.records.DAGProtos;
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
+import org.xerial.snappy.SnappyInputStream;
+import org.xerial.snappy.SnappyOutputStream;
 
 /**
  * Utility methods for setting up a DAG. Has helpers for setting up log4j configuration, converting
@@ -77,8 +76,7 @@
   public static ByteString createByteStringFromConf(Configuration conf) throws IOException {
     Preconditions.checkNotNull(conf, "Configuration must be specified");
     ByteString.Output os = ByteString.newOutput();
-    DeflaterOutputStream compressOs = new DeflaterOutputStream(os,
-        new Deflater(Deflater.BEST_SPEED));
+    SnappyOutputStream compressOs = new SnappyOutputStream(os);
     try {
       writeConfInPB(compressOs, conf);
     } finally {
@@ -111,9 +109,7 @@
    */
   public static Configuration createConfFromByteString(ByteString byteString) throws IOException {
     Preconditions.checkNotNull(byteString, "ByteString must be specified");
-    // SnappyInputStream uncompressIs = new
-    // SnappyInputStream(byteString.newInput());
-    try(InflaterInputStream uncompressIs = new InflaterInputStream(byteString.newInput())) {
+    try(SnappyInputStream uncompressIs = new SnappyInputStream(byteString.newInput());) {
       DAGProtos.ConfigurationProto confProto = DAGProtos.ConfigurationProto.parseFrom(uncompressIs);
       Configuration conf = new Configuration(false);
       readConfFromPB(confProto, conf);