Merge fixes.

20 files changed, 102 insertions, 68 deletions

diff --git a/common/src/main/java/org/apache/drill/common/logical/data/Flatten.java b/common/src/main/java/org/apache/drill/common/logical/data/Flatten.java
index dce509f23..07485d68c 100644
--- a/common/src/main/java/org/apache/drill/common/logical/data/Flatten.java
+++ b/common/src/main/java/org/apache/drill/common/logical/data/Flatten.java
@@ -29,13 +29,14 @@ import org.apache.drill.common.logical.data.visitors.LogicalVisitor;
 import java.util.Iterator;
 
 @JsonTypeName("flatten")
-public class Flatten extends SingleInputOperator{
+public class Flatten extends SingleInputOperator {
   private final FieldReference name;
   private final LogicalExpression expr;
   private final boolean drop;
 
   @JsonCreator
-  public Flatten(@JsonProperty("name") FieldReference name, @JsonProperty("expr") LogicalExpression expr, @JsonProperty("drop") boolean drop) {
+  public Flatten(@JsonProperty("name") FieldReference name, @JsonProperty("expr") LogicalExpression expr,
+      @JsonProperty("drop") boolean drop) {
     this.name = name;
     this.expr = expr;
     this.drop = drop;
@@ -53,17 +54,14 @@ public class Flatten extends SingleInputOperator{
     return drop;
   }
 
-    @Override
-    public <T, X, E extends Throwable> T accept(LogicalVisitor<T, X, E> logicalVisitor, X value) throws E {
-        return logicalVisitor.visitFlatten(this, value);
-    }
+  @Override
+  public <T, X, E extends Throwable> T accept(LogicalVisitor<T, X, E> logicalVisitor, X value) throws E {
+    return logicalVisitor.visitFlatten(this, value);
+  }
 
-    @Override
-    public Iterator<LogicalOperator> iterator() {
-        return Iterators.singletonIterator(getInput());
-    }
-  
-  
+  @Override
+  public Iterator<LogicalOperator> iterator() {
+    return Iterators.singletonIterator(getInput());
+  }
 
-	
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/infinispan/ICache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/infinispan/ICache.java
index d3b63db65..6627a892e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/infinispan/ICache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/infinispan/ICache.java
@@ -23,9 +23,11 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
-import com.google.common.collect.Maps;
 import org.apache.drill.common.config.DrillConfig;
 import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.cache.Counter;
@@ -55,6 +57,8 @@ import org.jgroups.protocols.COUNTER;
 import org.jgroups.protocols.FRAG2;
 import org.jgroups.stack.ProtocolStack;
 
+import com.google.common.collect.Maps;
+
 
 public class ICache implements DistributedCache{
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ICache.class);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
index 31ab9097d..7328257d9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/local/LocalCache.java
@@ -29,7 +29,11 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.*;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.drill.common.config.DrillConfig;
@@ -42,8 +46,6 @@ import org.apache.drill.exec.cache.DrillSerializable;
 import org.apache.drill.exec.exception.DrillbitStartupException;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.memory.TopLevelAllocator;
-import org.apache.drill.exec.proto.BitControl.PlanFragment;
-import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Charsets;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/Flatten.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/Flatten.java
index 2bcaab3f6..d123d2b00 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/Flatten.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/Flatten.java
@@ -17,11 +17,9 @@
  */
 package org.apache.drill.exec.physical.config;
 
-import org.apache.drill.exec.physical.OperatorCost;
 import org.apache.drill.exec.physical.base.AbstractSingle;
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.physical.base.PhysicalVisitor;
-import org.apache.drill.exec.physical.base.Size;
 import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
 import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
 
@@ -44,16 +42,6 @@ public class Flatten extends AbstractSingle {
   }
 
   @Override
-  public OperatorCost getCost() {
-    return new OperatorCost(0, 0, 1000, child.getSize().getRecordCount());
-  }
-
-  @Override
-  public Size getSize() {
-    return child.getSize();
-  }
-
-  @Override
   protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
     return new Flatten(child);
   }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
index bc2cdb5a5..2a19ba751 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/WireRecordBatch.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.record.RawFragmentBatchProvider;
 import org.apache.drill.exec.record.RecordBatch;
 import org.apache.drill.exec.record.RecordBatchLoader;
 import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.VectorWrapper;
 import org.apache.drill.exec.record.WritableBatch;
 import org.apache.drill.exec.record.selection.SelectionVector2;
@@ -166,5 +167,5 @@ public class WireRecordBatch implements RecordBatch {
   public VectorContainer getOutgoingContainer() {
     throw new UnsupportedOperationException(String.format(" You should not call getOutgoingContainer() for class %s", this.getClass().getCanonicalName()));
   }
-  
+
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
index 7535dcc3f..23296fbcc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/partitionsender/PartitionSenderRootExec.java
@@ -220,7 +220,6 @@ public class PartitionSenderRootExec extends BaseRootExec {
               operator.getOppositeMajorFragmentId(),
               fieldId,
               WritableBatch.getBatchNoHVWrap(0, container, false));
-      tunnel.sendRecordBatch(statusHandler, writableBatch);
       stats.startWait();
       try {
         tunnel.sendRecordBatch(statusHandler, writableBatch);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/PlanningSet.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/PlanningSet.java
index 3164228b7..da4380f3e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/PlanningSet.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/PlanningSet.java
@@ -25,28 +25,43 @@ import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
 
 import com.google.common.collect.Maps;
 
-public class PlanningSet implements Iterable<Wrapper>{
+public class PlanningSet implements Iterable<Wrapper> {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanningSet.class);
-  
+
   private Map<Fragment, Wrapper> fragmentMap = Maps.newHashMap();
+  private int majorFragmentIdIndex = 0;
 
-  PlanningSet(){
+  PlanningSet() {
   }
 
-  public void addAffinity(Fragment n, DrillbitEndpoint endpoint, float affinity){
+  public void addAffinity(Fragment n, DrillbitEndpoint endpoint, float affinity) {
     get(n).addEndpointAffinity(endpoint, affinity);
   }
-  
-  public void setWidth(Fragment n, int width){
+
+  public void setWidth(Fragment n, int width) {
     get(n).setWidth(width);
   }
-  
-  Wrapper get(Fragment node){
+
+  Wrapper get(Fragment node) {
     Wrapper wrapper = fragmentMap.get(node);
-    if(wrapper == null){
-      int majorFragmentId = node.getRoot().getOperatorId() == 0 ? 0 : node.getSendingExchange().getChild().getOperatorId() >> 16;
+    if (wrapper == null) {
+
+      int majorFragmentId = 0;
+
+      // If there is a sending exchange, we need to number other than zero.
+      if (node.getSendingExchange() != null) {
+
+        // assign the upper 16 bits as the major fragment id.
+        majorFragmentId = node.getSendingExchange().getChild().getOperatorId() >> 16;
+
+        // if they are not assigned, that means we mostly likely have an externally generated plan.  in this case, come up with a major fragmentid.
+        if (majorFragmentId == 0) {
+          majorFragmentId = majorFragmentIdIndex;
+        }
+      }
       wrapper = new Wrapper(node, majorFragmentId);
-      fragmentMap.put(node,  wrapper);
+      fragmentMap.put(node, wrapper);
+      majorFragmentIdIndex++;
     }
     return wrapper;
   }
@@ -60,7 +75,5 @@ public class PlanningSet implements Iterable<Wrapper>{
   public String toString() {
     return "FragmentPlanningSet:\n" + fragmentMap.values() + "]";
   }
-  
-  
-  
+
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FlattenPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FlattenPrel.java
index f54142260..3c19bae46 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FlattenPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/FlattenPrel.java
@@ -35,8 +35,7 @@ public class FlattenPrel extends SingleRel implements Prel {
   @Override
   public PhysicalOperator getPhysicalOperator(PhysicalPlanCreator creator) throws IOException {
     Flatten p = new Flatten(((Prel) getChild()).getPhysicalOperator(creator));
-    p.setOperatorId(creator.getOperatorId(this));
-    return p;
+    return creator.addMetadata(this, p);
   }
 
   @Override
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
index 0f0cf0a1e..017073546 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateTable.java
@@ -17,16 +17,17 @@
  */
 package org.apache.drill.exec.planner.sql.parser;
 
-import com.google.common.collect.Lists;
-import com.google.common.collect.ImmutableList;
+import java.util.List;
 
 import net.hydromatic.optiq.tools.Planner;
 
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.CreateTableHandler;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlIdentifier;
 import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlLiteral;
 import org.eigenbase.sql.SqlNode;
 import org.eigenbase.sql.SqlNodeList;
 import org.eigenbase.sql.SqlOperator;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateView.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateView.java
index 6f798b843..3666b4ccc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateView.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlCreateView.java
@@ -17,14 +17,14 @@
  */
 package org.apache.drill.exec.planner.sql.parser;
 
-import com.google.common.collect.Lists;
-import com.google.common.collect.ImmutableList;
+import java.util.List;
 
 import net.hydromatic.optiq.tools.Planner;
 
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.ViewHandler;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlIdentifier;
 import org.eigenbase.sql.SqlKind;
 import org.eigenbase.sql.SqlLiteral;
@@ -35,7 +35,8 @@ import org.eigenbase.sql.SqlSpecialOperator;
 import org.eigenbase.sql.SqlWriter;
 import org.eigenbase.sql.parser.SqlParserPos;
 
-import java.util.List;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
 
 public class SqlCreateView extends DrillSqlCall {
   public static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("CREATE_VIEW", SqlKind.OTHER){
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDescribeTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDescribeTable.java
index 8b022b093..15da82258 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDescribeTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDescribeTable.java
@@ -17,15 +17,17 @@
  */
 package org.apache.drill.exec.planner.sql.parser;
 
-import com.google.common.collect.Lists;
+import java.util.List;
 
 import net.hydromatic.optiq.tools.Planner;
 
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.DescribeTableHandler;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlIdentifier;
 import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlLiteral;
 import org.eigenbase.sql.SqlNode;
 import org.eigenbase.sql.SqlOperator;
 import org.eigenbase.sql.SqlSpecialOperator;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDropView.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDropView.java
index b86bd2083..1d943901a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDropView.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlDropView.java
@@ -17,6 +17,7 @@
  */
 package org.apache.drill.exec.planner.sql.parser;
 
+import java.util.Collections;
 import java.util.List;
 
 import net.hydromatic.optiq.tools.Planner;
@@ -24,8 +25,10 @@ import net.hydromatic.optiq.tools.Planner;
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.ViewHandler.DropView;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlIdentifier;
 import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlLiteral;
 import org.eigenbase.sql.SqlNode;
 import org.eigenbase.sql.SqlOperator;
 import org.eigenbase.sql.SqlSpecialOperator;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowFiles.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowFiles.java
index c929db636..690e5fd8c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowFiles.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowFiles.java
@@ -17,6 +17,7 @@
  */
 package org.apache.drill.exec.planner.sql.parser;
 
+import java.util.Collections;
 import java.util.List;
 
 import net.hydromatic.optiq.tools.Planner;
@@ -24,16 +25,16 @@ import net.hydromatic.optiq.tools.Planner;
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.ShowFileHandler;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlIdentifier;
 import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlLiteral;
 import org.eigenbase.sql.SqlNode;
 import org.eigenbase.sql.SqlOperator;
 import org.eigenbase.sql.SqlSpecialOperator;
 import org.eigenbase.sql.SqlWriter;
 import org.eigenbase.sql.parser.SqlParserPos;
 
-import com.google.common.collect.Lists;
-
 /**
  * Sql parse tree node to represent statement:
  * SHOW FILES [{FROM | IN} db_name] [LIKE 'pattern' | WHERE expr]
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowSchemas.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowSchemas.java
index 03e4e1dc1..faa69d107 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowSchemas.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowSchemas.java
@@ -24,7 +24,9 @@ import net.hydromatic.optiq.tools.Planner;
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.ShowSchemasHandler;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlLiteral;
 import org.eigenbase.sql.SqlNode;
 import org.eigenbase.sql.SqlOperator;
 import org.eigenbase.sql.SqlSpecialOperator;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowTables.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowTables.java
index ae9594404..e66199a6c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowTables.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlShowTables.java
@@ -24,8 +24,10 @@ import net.hydromatic.optiq.tools.Planner;
 import org.apache.drill.exec.ops.QueryContext;
 import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler;
 import org.apache.drill.exec.planner.sql.handlers.ShowTablesHandler;
+import org.eigenbase.sql.SqlCall;
 import org.eigenbase.sql.SqlIdentifier;
 import org.eigenbase.sql.SqlKind;
+import org.eigenbase.sql.SqlLiteral;
 import org.eigenbase.sql.SqlNode;
 import org.eigenbase.sql.SqlOperator;
 import org.eigenbase.sql.SqlSpecialOperator;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java
index 4ce5e98fb..5095d906d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java
@@ -17,7 +17,8 @@
  */
 package org.apache.drill.exec.store.hive;
 
-import com.beust.jcommander.internal.Maps;
+import java.util.Map;
+
 import org.apache.drill.exec.vector.NullableBigIntVector;
 import org.apache.drill.exec.vector.NullableBitVector;
 import org.apache.drill.exec.vector.NullableDateVector;
@@ -52,7 +53,7 @@ import org.apache.hadoop.io.Text;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
-import java.util.Map;
+import com.google.common.collect.Maps;
 
 public abstract class HiveFieldConverter {
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
index 1629f8c38..826a22f1a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java
@@ -339,10 +339,12 @@ public class Foreman implements Runnable, Closeable, Comparable<Object>{
       for(PhysicalOperator ops : plan.getSortedOperators()){
         size += ops.getCost();
       }
-      if(queuingEnabled && size > this.queueThreshold){
-        this.lease = largeSemaphore.acquire(this.queueTimeout, TimeUnit.MILLISECONDS);
-      }else{
-        this.lease = smallSemaphore.acquire(this.queueTimeout, TimeUnit.MILLISECONDS);
+      if(queuingEnabled){
+        if(size > this.queueThreshold){
+          this.lease = largeSemaphore.acquire(this.queueTimeout, TimeUnit.MILLISECONDS);
+        }else{
+          this.lease = smallSemaphore.acquire(this.queueTimeout, TimeUnit.MILLISECONDS);
+        }
       }
 
       QueryWorkUnit work = parallelizer.getFragments(context.getOptions().getSessionOptionList(), context.getCurrentEndpoint(),
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/cache/TestCacheSerialization.java b/exec/java-exec/src/test/java/org/apache/drill/exec/cache/TestCacheSerialization.java
index 16680b125..d548f6a49 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/cache/TestCacheSerialization.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/cache/TestCacheSerialization.java
@@ -60,10 +60,10 @@ public class TestCacheSerialization extends ExecTest {
   private static final DrillConfig CONFIG = DrillConfig.create();
 
   @Test
-  public void protobufSerialization() {
+  public void protobufSerialization() throws Exception {
     DistributedMap<String, FragmentHandle> map = ICACHE.getMap(CacheConfig.newBuilder(FragmentHandle.class).proto().build());
     FragmentHandle s = FragmentHandle.newBuilder().setMajorFragmentId(1).setMinorFragmentId(1).setQueryId(QueryId.newBuilder().setPart1(74).setPart2(66).build()).build();
-    map.put("1", s);
+    map.put("1", s).get();
     for(int i =0; i < 2; i++){
       FragmentHandle s2 = map.get("1");
       Assert.assertEquals(s, s2);
@@ -71,10 +71,10 @@ public class TestCacheSerialization extends ExecTest {
   }
 
   @Test
-  public void jacksonSerialization(){
+  public void jacksonSerialization() throws Exception {
     OptionValue v = OptionValue.createBoolean(OptionType.SESSION, "my test option", true);
     DistributedMap<String, OptionValue> map = ICACHE.getMap(CacheConfig.newBuilder(OptionValue.class).jackson().build());
-    map.put("1", v);
+    map.put("1", v).get();
     for(int i = 0; i < 5; i++){
       OptionValue v2 = map.get("1");
       Assert.assertEquals(v, v2);
@@ -114,7 +114,7 @@ public class TestCacheSerialization extends ExecTest {
     CachedVectorContainer wrap = new CachedVectorContainer(batch, ALLOCATOR);
 
     DistributedMultiMap<String, CachedVectorContainer> mmap = ICACHE.getMultiMap(OrderedPartitionRecordBatch.MULTI_CACHE_CONFIG);
-    mmap.put("vectors", wrap);
+    mmap.put("vectors", wrap).get();
 
     for(int x =0; x < 2; x++){
       CachedVectorContainer newWrap = (CachedVectorContainer) mmap.get("vectors").iterator().next();
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/SimpleRootExec.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/SimpleRootExec.java
index 7d0434023..7dce6e035 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/SimpleRootExec.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/SimpleRootExec.java
@@ -38,10 +38,11 @@ public class SimpleRootExec implements RootExec, Iterable<ValueVector>{
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SimpleRootExec.class);
 
   private RecordBatch incoming;
-
+  private ScreenRoot screenRoot;
   public SimpleRootExec(RootExec e){
     if(e instanceof ScreenRoot){
       incoming = ((ScreenRoot)e).getIncoming();
+      screenRoot = (ScreenRoot) e;
     }else{
       throw new UnsupportedOperationException();
     }
@@ -80,7 +81,7 @@ public class SimpleRootExec implements RootExec, Iterable<ValueVector>{
 
   @Override
   public void stop() {
-    incoming.cleanup();
+    screenRoot.stop();
   }
 
   @Override
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
index 702edb62e..73b7b53e3 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/TestCastFunctions.java
@@ -110,6 +110,8 @@ public class TestCastFunctions extends PopUnitTestBase{
       assertEquals(5, count);
     }
 
+    exec.stop();
+
     context.close();
     allocator.close();
 
@@ -155,6 +157,8 @@ public class TestCastFunctions extends PopUnitTestBase{
       assertEquals(5, count);
     }
 
+    exec.stop();
+
     context.close();
     allocator.close();
 
@@ -198,6 +202,8 @@ public class TestCastFunctions extends PopUnitTestBase{
       assertEquals(5, count);
     }
 
+    exec.stop();
+
     context.close();
     allocator.close();
 
@@ -242,6 +248,8 @@ public class TestCastFunctions extends PopUnitTestBase{
       assertEquals(5, count);
     }
 
+    exec.stop();
+
     context.close();
     allocator.close();
 
@@ -286,6 +294,8 @@ public class TestCastFunctions extends PopUnitTestBase{
       assertEquals(5, count);
     }
 
+    exec.stop();
+
     context.close();
     allocator.close();
 
@@ -330,6 +340,7 @@ public class TestCastFunctions extends PopUnitTestBase{
       }
       assertEquals(5, count);
     }
+    exec.stop();
 
     context.close();
     allocator.close();
@@ -374,6 +385,7 @@ public class TestCastFunctions extends PopUnitTestBase{
       }
       assertEquals(5, count);
     }
+    exec.stop();
 
     context.close();
     allocator.close();
@@ -408,6 +420,8 @@ public class TestCastFunctions extends PopUnitTestBase{
     while(exec.next()){
     }
 
+    exec.stop();
+
     context.close();
     allocator.close();