You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by rb...@apache.org on 2017/07/08 03:10:42 UTC
tez git commit: TEZ-3784. Submitting very large DAG throws
com.google.protobuf.CodedInputStream exception (rbalamohan)
Repository: tez
Updated Branches:
refs/heads/master b915a0772 -> 3d665f90c
TEZ-3784. Submitting very large DAG throws com.google.protobuf.CodedInputStream exception (rbalamohan)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/3d665f90
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/3d665f90
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/3d665f90
Branch: refs/heads/master
Commit: 3d665f90c16ede5bea7e046a31dac00f032d9e88
Parents: b915a07
Author: Rajesh Balamohan <rb...@apache.org>
Authored: Sat Jul 8 08:40:27 2017 +0530
Committer: Rajesh Balamohan <rb...@apache.org>
Committed: Sat Jul 8 08:40:27 2017 +0530
----------------------------------------------------------------------
.../client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java | 6 +++++-
.../rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java | 5 +++--
2 files changed, 8 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/3d665f90/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java
index baac186..72cf0d5 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java
@@ -23,6 +23,7 @@ import java.security.AccessControlException;
import java.util.List;
import java.util.Map;
+import com.google.protobuf.CodedInputStream;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -166,7 +167,10 @@ public class DAGClientAMProtocolBlockingPBServerImpl implements DAGClientAMProto
// need to deserialize large request from hdfs
Path requestPath = new Path(request.getSerializedRequestPath());
try (FSDataInputStream fsDataInputStream = stagingFs.open(requestPath)) {
- request = SubmitDAGRequestProto.parseFrom(fsDataInputStream);
+ CodedInputStream in =
+ CodedInputStream.newInstance(fsDataInputStream);
+ in.setSizeLimit(Integer.MAX_VALUE);
+ request = SubmitDAGRequestProto.parseFrom(in);
} catch (IOException e) {
throw wrapException(e);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/3d665f90/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java
index 3c030b3..040ca2f 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClientAMProtocolBlockingPBServerImpl.java
@@ -68,7 +68,7 @@ public class TestDAGClientAMProtocolBlockingPBServerImpl {
MockitoAnnotations.initMocks(this);
}
- @Test(timeout = 5000)
+ @Test(timeout = 100000)
@SuppressWarnings("unchecked")
public void testSubmitDagInSessionWithLargeDagPlan() throws Exception {
int maxIPCMsgSize = 1024;
@@ -77,7 +77,8 @@ public class TestDAGClientAMProtocolBlockingPBServerImpl {
TezConfiguration conf = new TezConfiguration();
conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, maxIPCMsgSize);
- byte[] randomBytes = new byte[2*maxIPCMsgSize];
+ // Check with 70 MB (64 MB is CodedInputStream's default limit in earlier versions of protobuf)
+ byte[] randomBytes = new byte[70 << 20];
(new Random()).nextBytes(randomBytes);
UserPayload payload = UserPayload.create(ByteBuffer.wrap(randomBytes));
Vertex vertex = Vertex.create("V", ProcessorDescriptor.create("P").setUserPayload(payload), 1);