apex-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From t..@apache.org
Subject [05/16] incubator-apex-core git commit: APEXCORE-423 style fix for apex-core engine
Date Sat, 09 Apr 2016 03:52:23 GMT
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java b/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
index 52ab342..17872cc 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/asm/CompactUtil.java
@@ -47,13 +47,13 @@ public class CompactUtil
     ccn.setAccess(cn.access);
     ccn.setDefaultConstructor(compactMethodNode(ASMUtil.getPublicDefaultConstructor(cn)));
     
-    List<CompactMethodNode> cmns = new LinkedList<CompactMethodNode>();
+    List<CompactMethodNode> cmns = new LinkedList<>();
     for (MethodNode mn : ASMUtil.getPublicGetter(cn)) {
       cmns.add(compactMethodNode(mn));
     }
     ccn.setGetterMethods(cmns);
     
-    cmns = new LinkedList<CompactMethodNode>();
+    cmns = new LinkedList<>();
     for (MethodNode mn : ASMUtil.getPublicSetter(cn)) {
       cmns.add(compactMethodNode(mn));
     }
@@ -62,7 +62,7 @@ public class CompactUtil
     ccn.setPorts(new LinkedList<CompactFieldNode>());
     ccn.setName(cn.name);
     
-    List<CompactClassNode> ccns = new LinkedList<CompactClassNode>();
+    List<CompactClassNode> ccns = new LinkedList<>();
     for (Object icn : cn.innerClasses) {
       CompactClassNode inner = new CompactClassNode();
       inner.setName(((InnerClassNode)icn).name);
@@ -72,8 +72,8 @@ public class CompactUtil
     if (ASMUtil.isEnum(cn)) {
       ccn.setEnumValues(ASMUtil.getEnumValues(cn));
     }
-    
-    if (cn instanceof ClassNodeType){
+
+    if (cn instanceof ClassNodeType) {
       ccn.setCsv(((ClassNodeType)cn).csv);
     }
     
@@ -95,8 +95,8 @@ public class CompactUtil
 
   public static void updateCompactClassPortInfo(ClassNode cn, CompactClassNode ccn)
   {
-    List<FieldNode> fields =  ASMUtil.getPorts(cn);
-    List<CompactFieldNode> ports = new LinkedList<CompactFieldNode>();
+    List<FieldNode> fields = ASMUtil.getPorts(cn);
+    List<CompactFieldNode> ports = new LinkedList<>();
     for (FieldNode fn : fields) {
       ports.add(compactFieldNode(fn));
     }
@@ -110,12 +110,14 @@ public class CompactUtil
     }
     CompactMethodNode cmn = new CompactMethodNode();
     cmn.setName(mn.name);
-    if(mn instanceof com.datatorrent.stram.webapp.asm.MethodNode)
-    cmn.setMethodSignatureNode(((com.datatorrent.stram.webapp.asm.MethodNode)mn).signatureNode);
+    if (mn instanceof com.datatorrent.stram.webapp.asm.MethodNode) {
+      cmn.setMethodSignatureNode(((com.datatorrent.stram.webapp.asm.MethodNode)mn).signatureNode);
+    }
     return cmn;
   }
 
-  private static CompactFieldNode compactFieldNode(FieldNode fn) {
+  private static CompactFieldNode compactFieldNode(FieldNode fn)
+  {
     if (fn == null) {
       return null;
     }
@@ -141,8 +143,9 @@ public class CompactUtil
     return cfn;
   }
 
-  private static void setAnnotationNode(FieldNode fn, CompactFieldNode cfn) {
-    List<CompactAnnotationNode> annotations = new LinkedList<CompactAnnotationNode>();
+  private static void setAnnotationNode(FieldNode fn, CompactFieldNode cfn)
+  {
+    List<CompactAnnotationNode> annotations = new LinkedList<>();
     for (Object visibleAnnotation : fn.visibleAnnotations) {
       CompactAnnotationNode node = new CompactAnnotationNode();
       Map<String, Object> annotationMap = new HashMap<String, Object>();

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/webapp/asm/FieldSignatureVisitor.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/asm/FieldSignatureVisitor.java b/engine/src/main/java/com/datatorrent/stram/webapp/asm/FieldSignatureVisitor.java
index 24b71d9..1e8420e 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/asm/FieldSignatureVisitor.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/asm/FieldSignatureVisitor.java
@@ -18,36 +18,39 @@
  */
 package com.datatorrent.stram.webapp.asm;
 
-
 import org.apache.xbean.asm5.signature.SignatureVisitor;
+
 /**
  * <p>FieldSignatureVisitor class.</p>
  *
  * @since 3.2.0
  */
-public class FieldSignatureVisitor extends BaseSignatureVisitor{
-
+public class FieldSignatureVisitor extends BaseSignatureVisitor
+{
   private Type fieldType;
-  
+
   @Override
-  public SignatureVisitor visitExceptionType() {
+  public SignatureVisitor visitExceptionType()
+  {
     throw new UnsupportedOperationException();
   }
 
   @Override
-  public SignatureVisitor visitParameterType() {
+  public SignatureVisitor visitParameterType()
+  {
     throw new UnsupportedOperationException();
   }
 
   @Override
-  public SignatureVisitor visitReturnType() {
+  public SignatureVisitor visitReturnType()
+  {
     throw new UnsupportedOperationException();
   }
 
-  public Type getFieldType() {
+  public Type getFieldType()
+  {
     if (!visitingStack.isEmpty()) {
       fieldType = visitingStack.pop();
-      ;
       visitingStack.push(fieldType);
       return fieldType;
     } else {

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodNode.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodNode.java b/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodNode.java
index da95ac2..a505260 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodNode.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodNode.java
@@ -45,8 +45,6 @@ public class MethodNode extends org.apache.xbean.asm5.tree.MethodNode
   {
     super.visitEnd();
     String methodString = signature != null ? signature : desc;
-    // System.out.println(methodString);
-    // System.out.println("RRRRRRRRRRRR" + methodString + "%%%%%" + name);
     SignatureReader reader = new SignatureReader(methodString);
     signatureNode = new MethodSignatureVisitor();
 //    signatureNode.signature = methodString;

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodSignatureVisitor.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodSignatureVisitor.java b/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodSignatureVisitor.java
index de9ccc6..3aa6d38 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodSignatureVisitor.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/asm/MethodSignatureVisitor.java
@@ -32,11 +32,11 @@ public class MethodSignatureVisitor extends BaseSignatureVisitor
 {
   
   // There is at most 1 parameter for setter and getter method
-  private List<Type> parameters = new LinkedList<Type>();
+  private List<Type> parameters = new LinkedList<>();
   
   private Type returnType;
   
-  private List<Type> exceptionType = new LinkedList<Type>();
+  private List<Type> exceptionType = new LinkedList<>();
   
   public static final int VISIT_PARAM = 1;
   
@@ -48,11 +48,10 @@ public class MethodSignatureVisitor extends BaseSignatureVisitor
   @Override
   public SignatureVisitor visitExceptionType()
   {
-    //System.out.print("visitExceptionType ");
-    if(stage == VISIT_RETURN && !visitingStack.isEmpty()){
+    if (stage == VISIT_RETURN && !visitingStack.isEmpty()) {
       returnType = visitingStack.pop();
     }
-    if(stage == VISIT_EXCEPTION && !visitingStack.isEmpty()){
+    if (stage == VISIT_EXCEPTION && !visitingStack.isEmpty()) {
       exceptionType.add(visitingStack.pop());
     }
     stage = VISIT_EXCEPTION;
@@ -63,17 +62,17 @@ public class MethodSignatureVisitor extends BaseSignatureVisitor
   @Override
   public SignatureVisitor visitParameterType()
   {
-    if(stage == VISIT_FORMAL_TYPE){
+    if (stage == VISIT_FORMAL_TYPE) {
       stage = VISIT_PARAM;
-      if(!visitingStack.isEmpty()){
+      if (!visitingStack.isEmpty()) {
         visitingStack.pop();
-      };
+      }
       return this;
-    } 
+    }
     stage = VISIT_PARAM;
-    if(!visitingStack.isEmpty()){
+    if (!visitingStack.isEmpty()) {
       parameters.add(visitingStack.pop());
-    };
+    }
     return this;
   }
   
@@ -82,7 +81,7 @@ public class MethodSignatureVisitor extends BaseSignatureVisitor
   public SignatureVisitor visitReturnType()
   {
 
-    while(!visitingStack.isEmpty()){
+    while (!visitingStack.isEmpty()) {
       parameters.add(visitingStack.pop());
     }
     stage = VISIT_RETURN;
@@ -92,7 +91,7 @@ public class MethodSignatureVisitor extends BaseSignatureVisitor
   
   public Type getReturnType()
   {
-    if(returnType==null && !visitingStack.isEmpty()){
+    if (returnType == null && !visitingStack.isEmpty()) {
       returnType = visitingStack.pop();
     }
     return returnType;
@@ -102,7 +101,5 @@ public class MethodSignatureVisitor extends BaseSignatureVisitor
   {
     return parameters;
   }
-  
-  
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/main/java/com/datatorrent/stram/webapp/asm/Type.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/webapp/asm/Type.java b/engine/src/main/java/com/datatorrent/stram/webapp/asm/Type.java
index 19c37c4..1e87b31 100644
--- a/engine/src/main/java/com/datatorrent/stram/webapp/asm/Type.java
+++ b/engine/src/main/java/com/datatorrent/stram/webapp/asm/Type.java
@@ -29,29 +29,28 @@ import org.apache.xbean.asm5.signature.SignatureVisitor;
  */
 public interface Type
 {
-
-  public static class TypeNode implements Type{
-    
+  class TypeNode implements Type
+  {
     private transient org.apache.xbean.asm5.Type typeObj;
-    
+
     private String objByteCode;
-    
+
     public org.apache.xbean.asm5.Type getTypeObj()
     {
-      if(typeObj == null){
+      if (typeObj == null) {
         typeObj = org.apache.xbean.asm5.Type.getType(objByteCode);
       }
       return typeObj;
     }
-    
+
     @Override
     public String toString()
     {
-      if(typeObj == null){
+      if (typeObj == null) {
         typeObj = org.apache.xbean.asm5.Type.getType(objByteCode);
       }
-      
-      if(typeObj.getSort()==org.apache.xbean.asm5.Type.OBJECT){
+
+      if (typeObj.getSort() == org.apache.xbean.asm5.Type.OBJECT) {
         return "class " + typeObj.getClassName();
       } else {
         return typeObj.getClassName();
@@ -61,7 +60,7 @@ public interface Type
     @Override
     public String getByteString()
     {
-      if(typeObj == null){
+      if (typeObj == null) {
         typeObj = org.apache.xbean.asm5.Type.getType(objByteCode);
       }
       if (typeObj.getSort() == org.apache.xbean.asm5.Type.OBJECT) {
@@ -80,34 +79,30 @@ public interface Type
     {
       this.objByteCode = objByteCode;
     }
-    
 
   }
-  
-  public static class WildcardTypeNode implements Type{
+
+  class WildcardTypeNode implements Type
+  {
 
     char boundChar;
-    
+
     ArrayList<Type> bounds = new ArrayList<Type>();
-    
+
     public Type[] getUpperBounds()
     {
-      if(boundChar == SignatureVisitor.EXTENDS)
-      {
+      if (boundChar == SignatureVisitor.EXTENDS) {
         return bounds.toArray(new Type[]{});
-      } else 
-      {
+      } else {
         return null;
       }
     }
 
     public Type[] getLowerBounds()
     {
-      if(boundChar == SignatureVisitor.SUPER)
-      {
+      if (boundChar == SignatureVisitor.SUPER) {
         return bounds.toArray(new Type[]{});
-      } else 
-      {
+      } else {
         return null;
       }
     }
@@ -117,59 +112,63 @@ public interface Type
     {
       return boundChar + "";
     }
-    
-    
-    
+
   }
-  
-  public static class TypeVariableNode implements Type {
+
+  class TypeVariableNode implements Type
+  {
 
     String typeLiteral;
-    
-    ArrayList<Type> bounds = new ArrayList<Type>();
-    
+
+    ArrayList<Type> bounds = new ArrayList<>();
+
     @Override
     public String getByteString()
     {
       return bounds.get(0).getByteString();
     }
-    
-    public Type[] getBounds() {
+
+    public Type[] getBounds()
+    {
       return bounds.toArray(new Type[]{});
     }
-    
+
     public String getTypeLiteral()
     {
       return typeLiteral;
     }
-    
-    public TypeNode getRawTypeBound(){
+
+    public TypeNode getRawTypeBound()
+    {
       Type t = bounds.get(0);
-      
+
       // The bounds can only be TypeNode or TypeVariableNode
-      if(t instanceof TypeNode){
-        return (TypeNode) t;
+      if (t instanceof TypeNode) {
+        return (TypeNode)t;
       }
       return ((TypeVariableNode)t).getRawTypeBound();
     }
-    
+
   }
-  
-  public static class ParameterizedTypeNode extends TypeNode {
-    
+
+  class ParameterizedTypeNode extends TypeNode
+  {
+
     ArrayList<Type> actualTypeArguments = new ArrayList<Type>();
-    
-    public Type[] getActualTypeArguments(){
+
+    public Type[] getActualTypeArguments()
+    {
       return actualTypeArguments.toArray(new Type[]{});
     }
   }
-  
-  
-  public static class ArrayTypeNode implements Type {
-    
+
+  class ArrayTypeNode implements Type
+  {
+
     Type actualArrayType;
-    
-    public Type getActualArrayType(){
+
+    public Type getActualArrayType()
+    {
       return actualArrayType;
     }
 
@@ -178,9 +177,9 @@ public interface Type
     {
       return "[" + actualArrayType.getByteString();
     }
-    
+
   }
-  
+
   String getByteString();
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/AffinityRulesTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/AffinityRulesTest.java b/engine/src/test/java/com/datatorrent/stram/AffinityRulesTest.java
index 73d5a1c..88bf133 100644
--- a/engine/src/test/java/com/datatorrent/stram/AffinityRulesTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/AffinityRulesTest.java
@@ -27,11 +27,12 @@ import java.util.Set;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.NodeState;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/GenericOperatorPropertyCodecTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/GenericOperatorPropertyCodecTest.java b/engine/src/test/java/com/datatorrent/stram/GenericOperatorPropertyCodecTest.java
index b2516fe..b1509e0 100644
--- a/engine/src/test/java/com/datatorrent/stram/GenericOperatorPropertyCodecTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/GenericOperatorPropertyCodecTest.java
@@ -18,8 +18,10 @@
  */
 package com.datatorrent.stram;
 
-import org.junit.Assert;
+import java.util.HashMap;
+import java.util.Map;
 
+import org.junit.Assert;
 import org.junit.Test;
 
 import com.datatorrent.api.StringCodec;
@@ -31,7 +33,6 @@ import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta;
 import com.datatorrent.stram.plan.physical.PhysicalPlan;
 import com.datatorrent.stram.plan.physical.PlanModifier;
 import com.datatorrent.stram.support.StramTestSupport.MemoryStorageAgent;
-import java.util.*;
 
 public class GenericOperatorPropertyCodecTest
 {

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/HostLocalTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/HostLocalTest.java b/engine/src/test/java/com/datatorrent/stram/HostLocalTest.java
index 184bf08..a2c58cb 100644
--- a/engine/src/test/java/com/datatorrent/stram/HostLocalTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/HostLocalTest.java
@@ -18,18 +18,18 @@
  */
 package com.datatorrent.stram;
 
-
 import java.io.File;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 
+import org.junit.Assert;
+import org.junit.Test;
+
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.NodeState;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
-import org.junit.Assert;
-import org.junit.Test;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -37,7 +37,6 @@ import com.google.common.collect.Sets;
 
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.api.DAG.Locality;
-
 import com.datatorrent.common.partitioner.StatelessPartitioner;
 import com.datatorrent.stram.StreamingContainerAgent.ContainerStartRequest;
 import com.datatorrent.stram.engine.GenericTestOperator;
@@ -55,7 +54,7 @@ public class HostLocalTest
     {
       Collection<Partition<GenericTestOperator>> newPartitions = super.definePartitions(partitions, context);
       Iterator<Partition<GenericTestOperator>> it = newPartitions.iterator();
-      for (int i=0; i<newPartitions.size() && it.hasNext(); i++) {
+      for (int i = 0; i < newPartitions.size() && it.hasNext(); i++) {
         it.next().getAttributes().put(OperatorContext.LOCALITY_HOST, "host" + (i + 1));
       }
       return newPartitions;
@@ -84,21 +83,21 @@ public class HostLocalTest
 
     int containerMem = 1000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
-    for(int i = 0; i< partitionCount; i++) {
-      NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host" + (i+1), 0),
-        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+    for (int i = 0; i < partitionCount; i++) {
+      NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host" + (i + 1), 0),
+          NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
       nodeReports.put(nr.getNodeId().getHost(), nr);
     }
 
     // set resources
     rr.updateNodeReports(Lists.newArrayList(nodeReports.values()));
     Set<String> expectedHosts = Sets.newHashSet();
-    for(int i =0; i< partitionCount; i++){
-      expectedHosts.add("host"+(i+1));
+    for (int i = 0; i < partitionCount; i++) {
+      expectedHosts.add("host" + (i + 1));
     }
     for (ContainerStartRequest csr : scm.containerStartRequests) {
       String host = rr.getHost(csr, true);
-      if(host != null){
+      if (host != null) {
         expectedHosts.remove(host);
       }
     }
@@ -130,10 +129,10 @@ public class HostLocalTest
     int containerMem = 1000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
     NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0),
-                                               NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
     nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0),
-                                    NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
 
     // set resources
@@ -169,10 +168,10 @@ public class HostLocalTest
     int containerMem = 1000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
     NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0),
-      NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
     nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0),
-      NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
 
     // set resources
@@ -206,11 +205,9 @@ public class HostLocalTest
 
     int containerMem = 1000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
-    NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0),
-                                               NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+    NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0), NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
-    nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0),
-                                    NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+    nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0), NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
 
     // set resources
@@ -246,10 +243,10 @@ public class HostLocalTest
     int containerMem = 1000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
     NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0),
-      NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
     nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0),
-      NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
 
     // set resources
@@ -275,9 +272,9 @@ public class HostLocalTest
 
     GenericTestOperator partitioned = dag.addOperator("partitioned", GenericTestOperator.class);
     dag.addStream("o1_outport1", o1.outport1, partitioned.inport1).setLocality(Locality.CONTAINER_LOCAL);
-    dag.setAttribute(o1,OperatorContext.MEMORY_MB,256);
-    dag.setAttribute(o1,OperatorContext.VCORES,2);
-    dag.setAttribute(partitioned,OperatorContext.VCORES,1);
+    dag.setAttribute(o1, OperatorContext.MEMORY_MB, 256);
+    dag.setAttribute(o1, OperatorContext.VCORES, 2);
+    dag.setAttribute(partitioned, OperatorContext.VCORES, 1);
 
     StreamingContainerManager scm = new StreamingContainerManager(dag);
 
@@ -286,10 +283,10 @@ public class HostLocalTest
     int containerMem = 1000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
     NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0),
-      NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
     nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0),
-      NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
+        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
 
     // set resources

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/InlineAM.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/InlineAM.java b/engine/src/test/java/com/datatorrent/stram/InlineAM.java
index cb473c5..3601c7e 100644
--- a/engine/src/test/java/com/datatorrent/stram/InlineAM.java
+++ b/engine/src/test/java/com/datatorrent/stram/InlineAM.java
@@ -45,7 +45,8 @@ import org.apache.hadoop.yarn.util.Records;
  * RM and negotiates a new attempt id. Then it waits for the RM app state to
  * reach be YarnApplicationState.ACCEPTED after which it runs the AM logic.
  */
-public abstract class InlineAM {
+public abstract class InlineAM
+{
   private static final Log LOG = LogFactory.getLog(InlineAM.class);
 
   // Handle to talk to the Resource Manager/Applications Manager
@@ -60,7 +61,8 @@ public abstract class InlineAM {
 
   /**
    */
-  public InlineAM(Configuration conf) throws Exception {
+  public InlineAM(Configuration conf) throws Exception
+  {
 
     appName = "UnmanagedAM";
     amPriority = 0;
@@ -73,7 +75,8 @@ public abstract class InlineAM {
 
   public abstract void runAM(ApplicationAttemptId attemptId) throws Exception;
 
-  public boolean run() throws Exception {
+  public boolean run() throws Exception
+  {
     LOG.info("Starting Client");
 
     // Connect to ResourceManager
@@ -160,7 +163,8 @@ public abstract class InlineAM {
    * @throws YarnRemoteException
    */
   private ApplicationReport monitorApplication(ApplicationId appId,
-      Set<YarnApplicationState> finalState) throws YarnException, IOException {
+      Set<YarnApplicationState> finalState) throws YarnException, IOException
+  {
 
     while (true) {
 

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/LocalityTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/LocalityTest.java b/engine/src/test/java/com/datatorrent/stram/LocalityTest.java
index 2b73e42..71c17d4 100644
--- a/engine/src/test/java/com/datatorrent/stram/LocalityTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/LocalityTest.java
@@ -21,11 +21,12 @@ package com.datatorrent.stram;
 import java.io.File;
 import java.util.Map;
 
+import org.junit.Assert;
+import org.junit.Test;
+
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.NodeState;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
-import org.junit.Assert;
-import org.junit.Test;
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -34,7 +35,6 @@ import com.google.common.collect.Sets;
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.api.Context.PortContext;
 import com.datatorrent.api.DAG.Locality;
-
 import com.datatorrent.common.partitioner.StatelessPartitioner;
 import com.datatorrent.stram.StreamingContainerAgent.ContainerStartRequest;
 import com.datatorrent.stram.engine.GenericTestOperator;
@@ -43,10 +43,12 @@ import com.datatorrent.stram.plan.physical.PTContainer;
 import com.datatorrent.stram.plan.physical.PTOperator;
 import com.datatorrent.stram.support.StramTestSupport.MemoryStorageAgent;
 
-public class LocalityTest {
+public class LocalityTest
+{
 
   @Test
-  public void testNodeLocal() {
+  public void testNodeLocal()
+  {
 
     LogicalPlan dag = new LogicalPlan();
     dag.getAttributes().put(com.datatorrent.api.Context.DAGContext.APPLICATION_PATH, new File("target", LocalityTest.class.getName()).getAbsolutePath());
@@ -77,11 +79,11 @@ public class LocalityTest {
 
     int containerMem = 2000;
     Map<String, NodeReport> nodeReports = Maps.newHashMap();
-    NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0),
-    		NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem*2, 2), 0, null, 0);
+    NodeReport nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host1", 0), NodeState.RUNNING, "httpAddress",
+        "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
-    nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0),
-        NodeState.RUNNING, "httpAddress", "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem*2, 2), 0, null, 0);
+    nr = BuilderUtils.newNodeReport(BuilderUtils.newNodeId("host2", 0), NodeState.RUNNING, "httpAddress",
+        "rackName", BuilderUtils.newResource(0, 0), BuilderUtils.newResource(containerMem * 2, 2), 0, null, 0);
     nodeReports.put(nr.getNodeId().getHost(), nr);
 
     // set resources
@@ -106,7 +108,7 @@ public class LocalityTest {
         if (oper.getNodeLocalOperators().getOperatorSet().size() > 1) {
           String expHost = null;
           for (PTOperator nodeLocalOper : oper.getNodeLocalOperators().getOperatorSet()) {
-            Assert.assertNotNull("host null "+nodeLocalOper.getContainer(), nodeLocalOper.getContainer().host);
+            Assert.assertNotNull("host null " + nodeLocalOper.getContainer(), nodeLocalOper.getContainer().host);
             if (expHost == null) {
               expHost = nodeLocalOper.getContainer().host;
             } else {

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java b/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
index daace0f..1704e55 100644
--- a/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/LogicalPlanModificationTest.java
@@ -31,11 +31,13 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 
+import com.google.common.collect.Sets;
+
+import com.datatorrent.api.AffinityRule;
 import com.datatorrent.api.AffinityRule.Type;
+import com.datatorrent.api.AffinityRulesSet;
 import com.datatorrent.api.Context.DAGContext;
 import com.datatorrent.api.DAG.Locality;
-import com.datatorrent.api.AffinityRule;
-import com.datatorrent.api.AffinityRulesSet;
 import com.datatorrent.api.StorageAgent;
 import com.datatorrent.common.util.AsyncFSStorageAgent;
 import com.datatorrent.common.util.FSStorageAgent;
@@ -43,9 +45,9 @@ import com.datatorrent.stram.engine.GenericTestOperator;
 import com.datatorrent.stram.engine.OperatorContext;
 import com.datatorrent.stram.engine.TestGeneratorInputOperator;
 import com.datatorrent.stram.plan.TestPlanContext;
-import com.datatorrent.stram.plan.logical.requests.CreateOperatorRequest;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta;
+import com.datatorrent.stram.plan.logical.requests.CreateOperatorRequest;
 import com.datatorrent.stram.plan.logical.requests.LogicalPlanRequest;
 import com.datatorrent.stram.plan.physical.PTContainer;
 import com.datatorrent.stram.plan.physical.PTOperator;
@@ -54,7 +56,6 @@ import com.datatorrent.stram.plan.physical.PlanModifier;
 import com.datatorrent.stram.support.StramTestSupport;
 import com.datatorrent.stram.support.StramTestSupport.TestMeta;
 
-import com.google.common.collect.Sets;
 
 public class LogicalPlanModificationTest
 {
@@ -219,6 +220,7 @@ public class LogicalPlanModificationTest
       pm.removeOperator(o2Meta.getName());
       Assert.fail("validation error (connected output stream) expected");
     } catch (ValidationException ve) {
+      // all good
     }
 
     // remove output stream required before removing operator
@@ -238,6 +240,7 @@ public class LogicalPlanModificationTest
       plan.getOperators(o2Meta);
       Assert.fail("removed from physical plan: " + o2Meta);
     } catch (Exception e) {
+      // all good
     }
     Assert.assertEquals("containers " + plan.getContainers(), 3, plan.getContainers().size());
     Assert.assertEquals("physical operators " + plan.getAllOperators(), 3, plan.getAllOperators().size());
@@ -375,9 +378,9 @@ public class LogicalPlanModificationTest
 
     lpmf.get();
 
-    Assert.assertEquals(""+dnm.containerStartRequests, 1, dnm.containerStartRequests.size());
+    Assert.assertEquals("" + dnm.containerStartRequests, 1, dnm.containerStartRequests.size());
     PTContainer c = dnm.containerStartRequests.poll().container;
-    Assert.assertEquals("operators "+c, 1, c.getOperators().size());
+    Assert.assertEquals("operators " + c, 1, c.getOperators().size());
 
     int deployStatusCnt = 0;
     for (PTOperator oper : c.getOperators()) {

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/MockContainer.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/MockContainer.java b/engine/src/test/java/com/datatorrent/stram/MockContainer.java
index 5b685bd..bb4502c 100644
--- a/engine/src/test/java/com/datatorrent/stram/MockContainer.java
+++ b/engine/src/test/java/com/datatorrent/stram/MockContainer.java
@@ -27,7 +27,6 @@ import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
 import com.datatorrent.api.Stats.OperatorStats;
-
 import com.datatorrent.stram.api.Checkpoint;
 import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.ContainerHeartbeat;
 import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.ContainerHeartbeatResponse;
@@ -58,11 +57,13 @@ public class MockContainer
     Assert.assertEquals(PTContainer.State.ACTIVE, container.getState());
   }
 
-  private StreamingContainerAgent assignContainer(StreamingContainerManager scm, PTContainer c) {
+  private StreamingContainerAgent assignContainer(StreamingContainerManager scm, PTContainer c)
+  {
     c.setResourceRequestPriority(c.getId());
     String containerId = "container" + c.getId();
-    InetSocketAddress bufferServerAddress = InetSocketAddress.createUnresolved(containerId+"Host", 0);
-    return scm.assignContainer(new StreamingContainerManager.ContainerResource(c.getId(), containerId, "localhost", 1024,0, null), bufferServerAddress);
+    InetSocketAddress bufferServerAddress = InetSocketAddress.createUnresolved(containerId + "Host", 0);
+    return scm.assignContainer(new StreamingContainerManager.ContainerResource(c.getId(), containerId, "localhost",
+        1024, 0, null), bufferServerAddress);
   }
 
   public void deploy()
@@ -77,7 +78,7 @@ public class MockContainer
 
     ContainerHeartbeatResponse chr = sca.dnmgr.processHeartbeat(hb); // get deploy request
     Assert.assertNotNull(chr.deployRequest);
-    Assert.assertEquals(""+chr.deployRequest, container.getOperators().size(), chr.deployRequest.size());
+    Assert.assertEquals("" + chr.deployRequest, container.getOperators().size(), chr.deployRequest.size());
     Assert.assertEquals(PTContainer.State.ACTIVE, container.getState());
 
     for (PTOperator oper : container.getOperators()) {
@@ -113,7 +114,8 @@ public class MockContainer
     Assert.assertNull(chr.deployRequest);
   }
 
-  public MockOperatorStats stats(int operatorId) {
+  public MockOperatorStats stats(int operatorId)
+  {
     MockOperatorStats os = this.stats.get(operatorId);
     if (os == null) {
       os = new MockOperatorStats(operatorId);
@@ -134,21 +136,24 @@ public class MockContainer
       this.operatorId = operatorId;
     }
 
-    public MockOperatorStats deployState(OperatorHeartbeat.DeployState s) {
+    public MockOperatorStats deployState(OperatorHeartbeat.DeployState s)
+    {
       this.deployState = s;
       return this;
     }
 
-    public MockOperatorStats currentWindowId(long windowId) {
+    public MockOperatorStats currentWindowId(long windowId)
+    {
       this.currentWindowId = windowId;
       return this;
     }
 
-    public MockOperatorStats checkpointWindowId(long windowId) {
+    public MockOperatorStats checkpointWindowId(long windowId)
+    {
       this.checkpointWindowId = windowId;
       return this;
     }
 
   }
 
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java b/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
index 7581cc3..470a666 100644
--- a/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/OutputUnifiedTest.java
@@ -29,7 +29,6 @@ import org.junit.Test;
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.api.Context.PortContext;
 import com.datatorrent.api.Operator;
-
 import com.datatorrent.common.partitioner.StatelessPartitioner;
 import com.datatorrent.stram.api.OperatorDeployInfo;
 import com.datatorrent.stram.engine.GenericTestOperator;
@@ -170,8 +169,9 @@ public class OutputUnifiedTest
     }
   }
 
-  private static StreamingContainerAgent assignContainer(StreamingContainerManager scm, String containerId) {
-    return scm.assignContainer(new StreamingContainerManager.ContainerResource(0, containerId, "localhost", 1024, 0, null), InetSocketAddress.createUnresolved(containerId+"Host", 0));
+  private static StreamingContainerAgent assignContainer(StreamingContainerManager scm, String containerId)
+  {
+    return scm.assignContainer(new StreamingContainerManager.ContainerResource(0, containerId, "localhost", 1024, 0, null), InetSocketAddress.createUnresolved(containerId + "Host", 0));
   }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/PartitioningTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/PartitioningTest.java b/engine/src/test/java/com/datatorrent/stram/PartitioningTest.java
index 9c97ab0..26575a4 100644
--- a/engine/src/test/java/com/datatorrent/stram/PartitioningTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/PartitioningTest.java
@@ -18,10 +18,17 @@
  */
 package com.datatorrent.stram;
 
-import com.datatorrent.common.util.BaseOperator;
 import java.io.File;
 import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.junit.After;
@@ -30,13 +37,21 @@ import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import com.google.common.collect.Sets;
 
-import com.datatorrent.api.*;
+import com.datatorrent.api.Context;
 import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.api.StatsListener;
 import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
 import com.datatorrent.common.partitioner.StatelessPartitioner;
 import com.datatorrent.common.util.AsyncFSStorageAgent;
+import com.datatorrent.common.util.BaseOperator;
 import com.datatorrent.stram.StramLocalCluster.LocalStreamingContainer;
 import com.datatorrent.stram.api.Checkpoint;
 import com.datatorrent.stram.engine.Node;
@@ -197,7 +212,7 @@ public class PartitioningTest
   public static class PartitionLoadWatch implements StatsListener, java.io.Serializable
   {
     private static final long serialVersionUID = 1L;
-    final private static ThreadLocal<Map<Integer, Integer>> loadIndicators = new ThreadLocal<Map<Integer, Integer>>();
+    private static final ThreadLocal<Map<Integer, Integer>> loadIndicators = new ThreadLocal<>();
 
     @Override
     public Response processStats(BatchedOperatorStats status)
@@ -215,7 +230,8 @@ public class PartitioningTest
       return hbr;
     }
 
-    public static void put(PTOperator oper, int load) {
+    public static void put(PTOperator oper, int load)
+    {
       Map<Integer, Integer> m = loadIndicators.get();
       if (m == null) {
         loadIndicators.set(m = new ConcurrentHashMap<Integer, Integer>());
@@ -223,7 +239,8 @@ public class PartitioningTest
       m.put(oper.getId(), load);
     }
 
-    public static void remove(PTOperator oper) {
+    public static void remove(PTOperator oper)
+    {
       loadIndicators.get().remove(oper.getId());
     }
 
@@ -308,7 +325,7 @@ public class PartitioningTest
 
     for (PTContainer container : lc.dnmgr.getPhysicalPlan().getContainers()) {
       int memory = 0;
-      for(PTOperator operator: container.getOperators()){
+      for (PTOperator operator : container.getOperators()) {
         memory += operator.getBufferServerMemory();
         memory += operator.getOperatorMeta().getValue(OperatorContext.MEMORY_MB);
       }
@@ -415,7 +432,7 @@ public class PartitioningTest
 
       List<PTOperator> partitions = assertNumberPartitions(3, lc, dag.getMeta(input));
       Set<String> partProperties = new HashSet<String>();
-      for (PTOperator p: partitions) {
+      for (PTOperator p : partitions) {
         LocalStreamingContainer c = StramTestSupport.waitForActivation(lc, p);
         Map<Integer, Node<?>> nodeMap = c.getNodes();
         Assert.assertEquals("number operators " + nodeMap, 1, nodeMap.size());
@@ -428,7 +445,7 @@ public class PartitioningTest
         p.setRecoveryCheckpoint(checkpoint);
         AsyncFSStorageAgent agent = new AsyncFSStorageAgent(checkpointDir.getPath(), null);
         agent.save(inputDeployed, p.getId(), 10L);
-        agent.copyToHDFS(p.getId(), 10l);
+        agent.copyToHDFS(p.getId(), 10L);
 
       }
 

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
index 53d6f31..e620141 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramLocalClusterTest.java
@@ -25,7 +25,11 @@ import java.io.LineNumberReader;
 import java.util.Arrays;
 import java.util.Map;
 
-import org.junit.*;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -33,7 +37,12 @@ import com.datatorrent.common.util.AsyncFSStorageAgent;
 import com.datatorrent.stram.StramLocalCluster.LocalStreamingContainer;
 import com.datatorrent.stram.StramLocalCluster.MockComponentFactory;
 import com.datatorrent.stram.api.Checkpoint;
-import com.datatorrent.stram.engine.*;
+import com.datatorrent.stram.engine.GenericTestOperator;
+import com.datatorrent.stram.engine.Node;
+import com.datatorrent.stram.engine.OperatorContext;
+import com.datatorrent.stram.engine.TestGeneratorInputOperator;
+import com.datatorrent.stram.engine.TestOutputOperator;
+import com.datatorrent.stram.engine.WindowGenerator;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.plan.physical.PTOperator;
 import com.datatorrent.stram.support.ManualScheduledExecutorService;

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java b/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
index d5cb14f..e2bb362 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramMiniClusterTest.java
@@ -18,39 +18,67 @@
  */
 package com.datatorrent.stram;
 
-import com.datatorrent.common.util.BaseOperator;
-import java.io.*;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
 import java.net.URL;
 import java.util.List;
 import java.util.Properties;
 
 import javax.ws.rs.core.MediaType;
 
+import org.codehaus.jettison.json.JSONObject;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.util.JarFinder;
 import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.*;
-import org.apache.hadoop.yarn.api.records.*;
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.NodeReport;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.client.api.AMRMClient;
 import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.MiniYARNCluster;
 import org.apache.hadoop.yarn.server.resourcemanager.ClientRMService;
 import org.apache.hadoop.yarn.util.Records;
-import org.codehaus.jettison.json.JSONObject;
-import org.junit.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Lists;
 import com.sun.jersey.api.client.Client;
 import com.sun.jersey.api.client.ClientResponse;
 import com.sun.jersey.api.client.WebResource;
 
-import com.datatorrent.api.*;
+import com.datatorrent.api.Context;
 import com.datatorrent.api.Context.OperatorContext;
-
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.common.util.BaseOperator;
 import com.datatorrent.stram.client.StramClientUtils;
 import com.datatorrent.stram.client.StramClientUtils.YarnClientHelper;
 import com.datatorrent.stram.engine.GenericTestOperator;
@@ -113,7 +141,7 @@ public class StramMiniClusterTest
 
     if (yarnCluster == null) {
       yarnCluster = new MiniYARNCluster(StramMiniClusterTest.class.getName(),
-                                        1, 1, 1);
+          1, 1, 1);
       yarnCluster.init(conf);
       yarnCluster.start();
     }
@@ -133,8 +161,7 @@ public class StramMiniClusterTest
 
     try {
       Thread.sleep(2000);
-    }
-    catch (InterruptedException e) {
+    } catch (InterruptedException e) {
       LOG.info("setup thread sleep interrupted. message=" + e.getMessage());
     }
   }
@@ -152,7 +179,7 @@ public class StramMiniClusterTest
   public void testSetupShutdown() throws Exception
   {
     GetClusterNodesRequest request =
-            Records.newRecord(GetClusterNodesRequest.class);
+        Records.newRecord(GetClusterNodesRequest.class);
     ClientRMService clientRMService = yarnCluster.getResourceManager().getClientRMService();
     GetClusterNodesResponse response = clientRMService.getClusterNodes(request);
     List<NodeReport> nodeReports = response.getNodeReports();
@@ -218,8 +245,7 @@ public class StramMiniClusterTest
 
       LOG.info("Client run completed. Result=" + result);
       Assert.assertTrue(result);
-    }
-    finally {
+    } finally {
       client.stop();
     }
   }
@@ -283,8 +309,7 @@ public class StramMiniClusterTest
       json = response.getEntity(JSONObject.class);
       LOG.info("Got response: " + json.toString());
 
-    }
-    finally {
+    } finally {
       //LOG.info("waiting...");
       //synchronized (this) {
       //  this.wait();
@@ -310,7 +335,7 @@ public class StramMiniClusterTest
       // Check maven pom.xml for generated classpath info
       // Works in tests where compile time env is same as runtime.
       ClassLoader thisClassLoader =
-              Thread.currentThread().getContextClassLoader();
+          Thread.currentThread().getContextClassLoader();
       String generatedClasspathFile = "mvn-generated-classpath";
       classpathFileStream =
               thisClassLoader.getResourceAsStream(generatedClasspathFile);
@@ -326,8 +351,7 @@ public class StramMiniClusterTest
       }
       // Put the file itself on classpath for tasks.
       envClassPath += thisClassLoader.getResource(generatedClasspathFile).getFile();
-    }
-    catch (IOException e) {
+    } catch (IOException e) {
       LOG.info("Could not find the necessary resource to generate class path for tests. Error=" + e.getMessage());
     }
 
@@ -338,8 +362,7 @@ public class StramMiniClusterTest
       if (reader != null) {
         reader.close();
       }
-    }
-    catch (IOException e) {
+    } catch (IOException e) {
       LOG.info("Failed to close class path file stream or reader. Error=" + e.getMessage());
     }
     return envClassPath;
@@ -381,11 +404,10 @@ public class StramMiniClusterTest
 
       ApplicationReport ar = client.getApplicationReport();
       Assert.assertEquals("should fail", FinalApplicationStatus.FAILED, ar.getFinalApplicationStatus());
-    // unable to get the diagnostics message set by the AM here - see YARN-208
+      // unable to get the diagnostics message set by the AM here - see YARN-208
       // diagnostics message does not make it here even with Hadoop 2.2 (but works on standalone cluster)
       //Assert.assertTrue("appReport " + ar, ar.getDiagnostics().contains("badOperator"));
-    }
-    finally {
+    } finally {
       client.stop();
     }
   }
@@ -400,12 +422,15 @@ public class StramMiniClusterTest
 
   @Ignore
   @Test
-  public void testUnmanagedAM() throws Exception {
+  public void testUnmanagedAM() throws Exception
+  {
 
-    new InlineAM(conf) {
+    new InlineAM(conf)
+    {
       @Override
       @SuppressWarnings("SleepWhileInLoop")
-      public void runAM(ApplicationAttemptId attemptId) throws Exception {
+      public void runAM(ApplicationAttemptId attemptId) throws Exception
+      {
         LOG.debug("AM running {}", attemptId);
 
         //AMRMClient amRmClient = new AMRMClientImpl(attemptId);
@@ -459,13 +484,13 @@ public class StramMiniClusterTest
         LOG.info("Requesting: " + req.getAskList());
         resourceManager.allocate(req);
 
-        for (int i=0; i<100; i++) {
+        for (int i = 0; i < 100; i++) {
           req = Records.newRecord(AllocateRequest.class);
           req.setResponseId(responseId++);
 
           AllocateResponse ar = resourceManager.allocate(req);
           sleep(1000);
-          LOG.debug("allocateResponse: {}" , ar);
+          LOG.debug("allocateResponse: {}", ar);
         }
 
         // unregister from RM
@@ -510,13 +535,16 @@ public class StramMiniClusterTest
 
   @Ignore
   @Test
-  public void testUnmanagedAM2() throws Exception {
+  public void testUnmanagedAM2() throws Exception
+  {
 
-    new InlineAM(conf) {
+    new InlineAM(conf)
+    {
 
       @Override
       @SuppressWarnings("SleepWhileInLoop")
-      public void runAM(ApplicationAttemptId attemptId) throws Exception {
+      public void runAM(ApplicationAttemptId attemptId) throws Exception
+      {
         LOG.debug("AM running {}", attemptId);
         AMRMClient<ContainerRequest> amRmClient = AMRMClient.createAMRMClient();
         amRmClient.init(conf);
@@ -548,10 +576,10 @@ public class StramMiniClusterTest
         req = new AMRMClient.ContainerRequest(capability, hosts, racks, priority, 3);
         amRmClient.addContainerRequest(req);
 */
-        for (int i=0; i<100; i++) {
+        for (int i = 0; i < 100; i++) {
           AllocateResponse ar = amRmClient.allocate(0);
           sleep(1000);
-          LOG.debug("allocateResponse: {}" , ar);
+          LOG.debug("allocateResponse: {}", ar);
           for (Container c : ar.getAllocatedContainers()) {
             LOG.debug("*** allocated {}", c.getResource());
             amRmClient.removeContainerRequest(req);

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java b/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
index 75b4684..0f56fa6 100644
--- a/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StramRecoveryTest.java
@@ -18,7 +18,14 @@
  */
 package com.datatorrent.stram;
 
-import java.io.*;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
 import java.net.URI;
@@ -26,15 +33,6 @@ import java.util.List;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.lang.mutable.MutableInt;
-import org.apache.commons.lang.mutable.MutableBoolean;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.ipc.RPC.Server;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.test.MockitoUtil;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
@@ -44,12 +42,20 @@ import org.mockito.invocation.InvocationOnMock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.commons.lang.mutable.MutableBoolean;
+import org.apache.commons.lang.mutable.MutableInt;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.RPC.Server;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.test.MockitoUtil;
+
 import com.google.common.collect.Lists;
 
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.api.StatsListener;
 import com.datatorrent.api.StorageAgent;
-
 import com.datatorrent.common.util.AsyncFSStorageAgent;
 import com.datatorrent.common.util.FSStorageAgent;
 import com.datatorrent.stram.api.Checkpoint;
@@ -210,7 +216,7 @@ public class StramRecoveryTest
 
     StreamingContainerAgent sca = scm.getContainerAgent(originalContainer.getExternalId());
     Assert.assertNotNull("allocated container restored " + originalContainer, sca);
-    assertEquals("memory usage allocated container", (int) OperatorContext.MEMORY_MB.defaultValue, sca.container.getAllocatedMemoryMB());
+    assertEquals("memory usage allocated container", (int)OperatorContext.MEMORY_MB.defaultValue, sca.container.getAllocatedMemoryMB());
 
     // YARN-1490 - simulate container terminated on AM recovery
     scm.scheduleContainerRestart(originalContainer.getExternalId());
@@ -294,14 +300,18 @@ public class StramRecoveryTest
     StreamingContainerManager scm = new StreamingContainerManager(dag);
     PhysicalPlan plan = scm.getPhysicalPlan();
     Journal j = scm.getJournal();
-    ByteArrayOutputStream bos = new ByteArrayOutputStream() {
+    ByteArrayOutputStream bos = new ByteArrayOutputStream()
+    {
       @Override
-      public void flush() throws IOException {
+      public void flush() throws IOException
+      {
         super.flush();
         flushCount.increment();
       }
+
       @Override
-      public void close() throws IOException {
+      public void close() throws IOException
+      {
         super.close();
         isClosed.setValue(true);
       }
@@ -424,8 +434,7 @@ public class StramRecoveryTest
     try {
       sc.start();
       sc.copyInitialState(new Path(appPath1));
-    }
-    finally {
+    } finally {
       sc.stop();
     }
     scm = StreamingContainerManager.getInstance(new FSRecoveryHandler(dag.assertAppPath(), new Configuration(false)), dag, false);
@@ -465,20 +474,24 @@ public class StramRecoveryTest
 
     StreamingContainerUmbilicalProtocol impl = MockitoUtil.mockProtocol(StreamingContainerUmbilicalProtocol.class);
 
-    Mockito.doAnswer(new org.mockito.stubbing.Answer<Void>() {
+    Mockito.doAnswer(new org.mockito.stubbing.Answer<Void>()
+    {
       @Override
-      public Void answer(InvocationOnMock invocation) {
+      public Void answer(InvocationOnMock invocation)
+      {
         LOG.debug("got call: " + invocation.getMethod());
         if (!timedout.get()) {
           try {
             timedout.set(true);
             Thread.sleep(1000);
           } catch (Exception e) {
+            // ignore
           }
           //throw new RuntimeException("fail");
         }
         return null;
-      }})
+      }
+    })
     .when(impl).log("containerId", "timeout");
 
     Server server = new RPC.Builder(conf).setProtocol(StreamingContainerUmbilicalProtocol.class).setInstance(impl)
@@ -552,7 +565,6 @@ public class StramRecoveryTest
     protocolProxy.log("containerId", "timeout");
     Assert.assertTrue("timedout", timedout.get());
 
-
     restoreSystemProperty(RecoverableRpcProxy.RPC_TIMEOUT, rpcTimeout);
     restoreSystemProperty(RecoverableRpcProxy.RETRY_DELAY, rpcRetryDelay);
     restoreSystemProperty(RecoverableRpcProxy.RETRY_TIMEOUT, rpcRetryTimeout);
@@ -562,7 +574,7 @@ public class StramRecoveryTest
 
   private static String restoreSystemProperty(final String key, final String value)
   {
-    return (value == null)? System.clearProperty(key) : System.setProperty(key, value);
+    return (value == null) ? System.clearProperty(key) : System.setProperty(key, value);
   }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java b/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
index ddf3448..d9a9ee4 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
@@ -18,8 +18,26 @@
  */
 package com.datatorrent.stram;
 
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
 
-import com.datatorrent.api.*;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.StatsListener;
+import com.datatorrent.api.StreamCodec;
 import com.datatorrent.api.annotation.InputPortFieldAnnotation;
 import com.datatorrent.common.partitioner.StatelessPartitioner;
 import com.datatorrent.stram.api.OperatorDeployInfo;
@@ -30,13 +48,6 @@ import com.datatorrent.stram.plan.physical.PTContainer;
 import com.datatorrent.stram.plan.physical.PTOperator;
 import com.datatorrent.stram.plan.physical.PhysicalPlan;
 import com.datatorrent.stram.support.StramTestSupport;
-import com.google.common.collect.Lists;
-import java.io.Serializable;
-import java.util.*;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
 
 /**
  */
@@ -292,7 +303,7 @@ public class StreamCodecTest
 
     for (PTContainer container : containers) {
       List<PTOperator> operators = container.getOperators();
-      for (PTOperator operator :operators) {
+      for (PTOperator operator : operators) {
         if (!operator.isUnifier()) {
           if (operator.getOperatorMeta() == n1meta) {
             OperatorDeployInfo odi = getOperatorDeployInfo(operator, n1meta.getName(), dnm);
@@ -348,7 +359,7 @@ public class StreamCodecTest
     List<PTContainer> containers = plan.getContainers();
 
     for (int i = 0; i < containers.size(); ++i) {
-      StreamingContainerManagerTest.assignContainer(dnm, "container" + (i+1));
+      StreamingContainerManagerTest.assignContainer(dnm, "container" + (i + 1));
     }
 
     LogicalPlan.OperatorMeta n1meta = dag.getMeta(node1);
@@ -495,7 +506,7 @@ public class StreamCodecTest
 
     for (PTContainer container : containers) {
       List<PTOperator> operators = container.getOperators();
-      for (PTOperator operator :operators) {
+      for (PTOperator operator : operators) {
         if (!operator.isUnifier()) {
           if (operator.getOperatorMeta() == n1meta) {
             OperatorDeployInfo odi = getOperatorDeployInfo(operator, n1meta.getName(), dnm);
@@ -626,7 +637,7 @@ public class StreamCodecTest
 
     for (PTContainer container : containers) {
       List<PTOperator> operators = container.getOperators();
-      for (PTOperator operator :operators) {
+      for (PTOperator operator : operators) {
         if (!operator.isUnifier()) {
           if (operator.getOperatorMeta() == n1meta) {
             OperatorDeployInfo odi = getOperatorDeployInfo(operator, n1meta.getName(), dnm);
@@ -689,7 +700,6 @@ public class StreamCodecTest
     TestStreamCodec serDe2 = new TestStreamCodec();
     dag.setInputPortAttribute(node3.inport1, Context.PortContext.STREAM_CODEC, serDe2);
 
-
     dag.addStream("n1n2n3", node1.outport1, node2.inport1, node3.inport1);
 
     dag.setAttribute(LogicalPlan.CONTAINERS_MAX_COUNT, Integer.MAX_VALUE);
@@ -727,7 +737,7 @@ public class StreamCodecTest
     LogicalPlan.OperatorMeta n3meta = dag.getMeta(node3);
     for (PTContainer container : containers) {
       List<PTOperator> operators = container.getOperators();
-      for (PTOperator operator :operators) {
+      for (PTOperator operator : operators) {
         if (!operator.isUnifier()) {
           if (operator.getOperatorMeta() == n1meta) {
             OperatorDeployInfo odi = getOperatorDeployInfo(operator, n1meta.getName(), dnm);
@@ -892,7 +902,7 @@ public class StreamCodecTest
 
     for (PTContainer container : containers) {
       List<PTOperator> operators = container.getOperators();
-      for (PTOperator operator :operators) {
+      for (PTOperator operator : operators) {
         if (!operator.isUnifier()) {
           if (operator.getOperatorMeta() == n1meta) {
             OperatorDeployInfo odi = getOperatorDeployInfo(operator, n1meta.getName(), dnm);
@@ -946,7 +956,7 @@ public class StreamCodecTest
   {
     GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
     dag.setAttribute(node1, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(2));
-    dag.setAttribute(node1, Context.OperatorContext.STATS_LISTENERS, Lists.newArrayList((StatsListener) new PartitioningTest.PartitionLoadWatch()));
+    dag.setAttribute(node1, Context.OperatorContext.STATS_LISTENERS, Lists.newArrayList((StatsListener)new PartitioningTest.PartitionLoadWatch()));
     GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
     dag.setAttribute(node2, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(3));
     dag.setAttribute(node2, Context.OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{new PartitioningTest.PartitionLoadWatch()}));
@@ -958,7 +968,6 @@ public class StreamCodecTest
     TestStreamCodec serDe2 = new TestStreamCodec();
     dag.setInputPortAttribute(node3.inport1, Context.PortContext.STREAM_CODEC, serDe2);
 
-
     dag.addStream("n1n2n3", node1.outport1, node2.inport1, node3.inport1);
 
     dag.setAttribute(LogicalPlan.CONTAINERS_MAX_COUNT, Integer.MAX_VALUE);
@@ -989,7 +998,7 @@ public class StreamCodecTest
     // scale down N (node2) from 3 to 2 and then from 2 to 1
     for (int i = 0; i < 2; i++) {
       markAllOperatorsActive(plan);
-      List<PTOperator> ptos =  plan.getOperators(n2meta);
+      List<PTOperator> ptos = plan.getOperators(n2meta);
       for (PTOperator ptOperator : ptos) {
         PartitioningTest.PartitionLoadWatch.put(ptOperator, -1);
         plan.onStatusUpdate(ptOperator);
@@ -1012,13 +1021,13 @@ public class StreamCodecTest
         */
       }
 
-      Assert.assertEquals("Number of unifiers ", 2-i, upstreamOperators.size());
+      Assert.assertEquals("Number of unifiers ", 2 - i, upstreamOperators.size());
     }
 
     // scale down N (node3) from 3 to 2 and then from 2 to 1
     for (int i = 0; i < 2; i++) {
       markAllOperatorsActive(plan);
-      List<PTOperator> ptos =  plan.getOperators(n3meta);
+      List<PTOperator> ptos = plan.getOperators(n3meta);
       for (PTOperator ptOperator : ptos) {
         PartitioningTest.PartitionLoadWatch.put(ptOperator, -1);
         plan.onStatusUpdate(ptOperator);
@@ -1033,7 +1042,7 @@ public class StreamCodecTest
         upstreamOperators.addAll(operator.upstreamMerge.values());
       }
 
-      Assert.assertEquals("Number of unifiers ", 2-i, upstreamOperators.size());
+      Assert.assertEquals("Number of unifiers ", 2 - i, upstreamOperators.size());
     }
 
     // Check that different unifiers were created for the two output operators with different codecs
@@ -1070,7 +1079,7 @@ public class StreamCodecTest
         */
       }
 
-      Assert.assertEquals("Number of unifiers ", 2+i, upstreamOperators.size());
+      Assert.assertEquals("Number of unifiers ", 2 + i, upstreamOperators.size());
     }
 
     // scale down M to 1
@@ -1125,7 +1134,6 @@ public class StreamCodecTest
     return lastId;
   }
 
-
   private void markAllOperatorsActive(PhysicalPlan plan)
   {
     for (PTContainer container : plan.getContainers()) {
@@ -1149,8 +1157,8 @@ public class StreamCodecTest
   }
 
   private void checkPresentStreamCodec(LogicalPlan.OperatorMeta operatorMeta, Operator.InputPort<?> inputPort,
-                                       Map<Integer, StreamCodec<?>> streamCodecs,
-                                       String id, PhysicalPlan plan )
+      Map<Integer, StreamCodec<?>> streamCodecs,
+      String id, PhysicalPlan plan)
   {
     StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(operatorMeta.getMeta(inputPort));
     Assert.assertTrue("stream codec identifier not present" + id, isStrCodecPresent(streamCodecInfo, plan));
@@ -1159,7 +1167,7 @@ public class StreamCodecTest
   }
 
   private void checkPresentStreamCodecInfo(Map<Integer, StreamCodec<?>> streamCodecs, String id,
-                                           Integer streamCodecIdentifier, StreamCodec<?> streamCodecInfo)
+      Integer streamCodecIdentifier, StreamCodec<?> streamCodecInfo)
   {
     StreamCodec<?> opStreamCodecInfo = streamCodecs.get(streamCodecIdentifier);
     Assert.assertNotNull("stream codec info null " + id, opStreamCodecInfo);
@@ -1193,11 +1201,12 @@ public class StreamCodecTest
       }
     }
 
-    Assert.assertNotNull(id + " assigned to " + containerId + " deploy info is null", odi );
+    Assert.assertNotNull(id + " assigned to " + containerId + " deploy info is null", odi);
     return odi;
   }
 
-  private OperatorDeployInfo.InputDeployInfo getInputDeployInfo(OperatorDeployInfo odi, LogicalPlan.InputPortMeta portMeta)
+  private OperatorDeployInfo.InputDeployInfo getInputDeployInfo(OperatorDeployInfo odi, LogicalPlan.InputPortMeta
+      portMeta)
   {
     OperatorDeployInfo.InputDeployInfo idi = null;
     List<OperatorDeployInfo.InputDeployInfo> inputs = odi.inputs;
@@ -1237,7 +1246,7 @@ public class StreamCodecTest
     @Override
     public int getPartition(Object o)
     {
-      return o.hashCode()/2;
+      return o.hashCode() / 2;
     }
   }
 
@@ -1247,11 +1256,11 @@ public class StreamCodecTest
     @Override
     public int getPartition(Object o)
     {
-      return o.hashCode()/3;
+      return o.hashCode() / 3;
     }
   }
 
-  public static class DefaultTestStreamCodec  extends DefaultStatefulStreamCodec<Object> implements Serializable
+  public static class DefaultTestStreamCodec extends DefaultStatefulStreamCodec<Object> implements Serializable
   {
     private static final long serialVersionUID = 1L;
   }
@@ -1261,7 +1270,8 @@ public class StreamCodecTest
     private static final DefaultTestStreamCodec codec = new DefaultTestStreamCodec();
 
     @InputPortFieldAnnotation(optional = true)
-    final public transient InputPort<Object> inportWithCodec = new DefaultInputPort<Object>() {
+    public final transient InputPort<Object> inportWithCodec = new DefaultInputPort<Object>()
+    {
       @Override
       public StreamCodec<Object> getStreamCodec()
       {
@@ -1269,7 +1279,7 @@ public class StreamCodecTest
       }
 
       @Override
-      final public void process(Object payload)
+      public final void process(Object payload)
       {
       }
 

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
index a5006bf..f29403a 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamingContainerManagerTest.java
@@ -21,17 +21,26 @@ package com.datatorrent.stram;
 import java.io.IOException;
 import java.io.Serializable;
 import java.net.InetSocketAddress;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.hadoop.io.DataInputByteBuffer;
-import org.apache.hadoop.io.DataOutputByteBuffer;
+import org.codehaus.jettison.json.JSONObject;
+import org.eclipse.jetty.websocket.WebSocket;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.io.DataInputByteBuffer;
+import org.apache.hadoop.io.DataOutputByteBuffer;
+
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 
@@ -64,7 +73,13 @@ import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.OperatorHea
 import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.OperatorHeartbeat.DeployState;
 import com.datatorrent.stram.appdata.AppDataPushAgent;
 import com.datatorrent.stram.codec.DefaultStatefulStreamCodec;
-import com.datatorrent.stram.engine.*;
+import com.datatorrent.stram.engine.DefaultUnifier;
+import com.datatorrent.stram.engine.GenericTestOperator;
+import com.datatorrent.stram.engine.TestAppDataQueryOperator;
+import com.datatorrent.stram.engine.TestAppDataResultOperator;
+import com.datatorrent.stram.engine.TestAppDataSourceOperator;
+import com.datatorrent.stram.engine.TestGeneratorInputOperator;
+import com.datatorrent.stram.engine.WindowGenerator;
 import com.datatorrent.stram.plan.TestPlanContext;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta;
@@ -80,10 +95,6 @@ import com.datatorrent.stram.support.StramTestSupport.TestMeta;
 import com.datatorrent.stram.tuple.Tuple;
 import com.datatorrent.stram.webapp.LogicalOperatorInfo;
 
-import org.apache.commons.lang.StringUtils;
-import org.codehaus.jettison.json.JSONObject;
-import org.eclipse.jetty.websocket.WebSocket;
-
 public class StreamingContainerManagerTest
 {
   @Rule
@@ -161,9 +172,9 @@ public class StreamingContainerManagerTest
     dag.setOutputPortAttribute(o1.outport, PortContext.SPIN_MILLIS, 99);
 
     dag.addStream("o2.outport1", o2.outport1, o3.inport1)
-      .setLocality(Locality.CONTAINER_LOCAL);
+        .setLocality(Locality.CONTAINER_LOCAL);
     dag.addStream("o3.outport1", o3.outport1, o4.inport1)
-      .setLocality(Locality.THREAD_LOCAL);
+        .setLocality(Locality.THREAD_LOCAL);
 
     dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 2);
     dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
@@ -277,8 +288,8 @@ public class StreamingContainerManagerTest
 
     Assert.assertEquals("number containers", 6, plan.getContainers().size());
     List<StreamingContainerAgent> containerAgents = Lists.newArrayList();
-    for (int i=0; i < plan.getContainers().size(); i++) {
-      containerAgents.add(assignContainer(dnm, "container"+(i+1)));
+    for (int i = 0; i < plan.getContainers().size(); i++) {
+      containerAgents.add(assignContainer(dnm, "container" + (i + 1)));
     }
 
     PTContainer c = plan.getOperators(dag.getMeta(node1)).get(0).getContainer();
@@ -290,7 +301,7 @@ public class StreamingContainerManagerTest
     List<PTOperator> o2Partitions = plan.getOperators(dag.getMeta(node2));
     Assert.assertEquals("number partitions", TestStaticPartitioningSerDe.partitions.length, o2Partitions.size());
 
-    for (int i=0; i<o2Partitions.size(); i++) {
+    for (int i = 0; i < o2Partitions.size(); i++) {
       String containerId = o2Partitions.get(i).getContainer().getExternalId();
       List<OperatorDeployInfo> cc = getDeployInfo(dnm.getContainerAgent(containerId));
       Assert.assertEquals("number operators assigned to container", 1, cc.size());
@@ -325,7 +336,7 @@ public class StreamingContainerManagerTest
       Assert.assertEquals("portName " + nidi, mergePortName, nidi.portName);
       Assert.assertNotNull("sourceNodeId " + nidi, nidi.sourceNodeId);
       Assert.assertNotNull("contextAttributes " + nidi, nidi.contextAttributes);
-      Assert.assertEquals("contextAttributes " , new Integer(1111), nidi.getValue(PortContext.QUEUE_CAPACITY));
+      Assert.assertEquals("contextAttributes ", new Integer(1111), nidi.getValue(PortContext.QUEUE_CAPACITY));
       sourceNodeIds.add(nidi.sourceNodeId);
     }
     for (PTOperator node : dnm.getPhysicalPlan().getOperators(dag.getMeta(node2))) {
@@ -335,14 +346,13 @@ public class StreamingContainerManagerTest
     Assert.assertEquals("outputs " + mergeNodeDI, 1, mergeNodeDI.outputs.size());
     for (OutputDeployInfo odi : mergeNodeDI.outputs) {
       Assert.assertNotNull("contextAttributes " + odi, odi.contextAttributes);
-      Assert.assertEquals("contextAttributes " , new Integer(2222), odi.getValue(PortContext.QUEUE_CAPACITY));
+      Assert.assertEquals("contextAttributes ", new Integer(2222), odi.getValue(PortContext.QUEUE_CAPACITY));
     }
 
     try {
       Object operator = msa.load(mergeNodeDI.id, Stateless.WINDOW_ID);
-      Assert.assertTrue("" + operator,  operator instanceof DefaultUnifier);
-    }
-    catch (IOException ex) {
+      Assert.assertTrue("" + operator, operator instanceof DefaultUnifier);
+    } catch (IOException ex) {
       throw new RuntimeException(ex);
     }
 
@@ -375,19 +385,19 @@ public class StreamingContainerManagerTest
     dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
 
     StreamingContainerManager scm = new StreamingContainerManager(dag);
-    Assert.assertEquals(""+scm.containerStartRequests, 2, scm.containerStartRequests.size());
+    Assert.assertEquals("" + scm.containerStartRequests, 2, scm.containerStartRequests.size());
     scm.containerStartRequests.clear();
 
     PhysicalPlan plan = scm.getPhysicalPlan();
 
     List<PTContainer> containers = plan.getContainers();
-    Assert.assertEquals(""+containers, 2, plan.getContainers().size());
+    Assert.assertEquals("" + containers, 2, plan.getContainers().size());
 
     PTContainer c1 = containers.get(0);
-    Assert.assertEquals("c1.operators "+c1.getOperators(), 2, c1.getOperators().size());
+    Assert.assertEquals("c1.operators " + c1.getOperators(), 2, c1.getOperators().size());
 
     PTContainer c2 = containers.get(1);
-    Assert.assertEquals("c2.operators "+c2.getOperators(), 1, c2.getOperators().size());
+    Assert.assertEquals("c2.operators " + c2.getOperators(), 1, c2.getOperators().size());
 
     assignContainer(scm, "container1");
     assignContainer(scm, "container2");
@@ -400,12 +410,12 @@ public class StreamingContainerManagerTest
     scm.scheduleContainerRestart(c1.getExternalId());
     Assert.assertEquals("", 0, countState(sca1.container, PTOperator.State.PENDING_UNDEPLOY));
     Assert.assertEquals("", 2, countState(sca1.container, PTOperator.State.PENDING_DEPLOY));
-    Assert.assertEquals(""+scm.containerStartRequests, 1, scm.containerStartRequests.size());
+    Assert.assertEquals("" + scm.containerStartRequests, 1, scm.containerStartRequests.size());
     ContainerStartRequest dr = scm.containerStartRequests.peek();
     Assert.assertNotNull(dr);
 
-    Assert.assertEquals(""+sca2.container, 1, countState(sca2.container, PTOperator.State.PENDING_UNDEPLOY));
-    Assert.assertEquals(""+sca2.container, 0, countState(sca2.container, PTOperator.State.PENDING_DEPLOY));
+    Assert.assertEquals("" + sca2.container, 1, countState(sca2.container, PTOperator.State.PENDING_UNDEPLOY));
+    Assert.assertEquals("" + sca2.container, 0, countState(sca2.container, PTOperator.State.PENDING_DEPLOY));
 
   }
 
@@ -467,8 +477,7 @@ public class StreamingContainerManagerTest
     try {
       sa.getWindowIds(1);
       Assert.fail("There should not be any most recently saved windowId!");
-    }
-    catch (IOException io) {
+    } catch (IOException io) {
       Assert.assertTrue("No State Saved", true);
     }
   }
@@ -527,7 +536,7 @@ public class StreamingContainerManagerTest
 
     ContainerHeartbeatResponse chr = scm.processHeartbeat(hb); // get deploy request
     Assert.assertNotNull(chr.deployRequest);
-    Assert.assertEquals(""+chr.deployRequest, 1, chr.deployRequest.size());
+    Assert.assertEquals("" + chr.deployRequest, 1, chr.deployRequest.size());
     Assert.assertEquals(PTContainer.State.ACTIVE, o1p1.getContainer().getState());
     Assert.assertEquals("state " + o1p1, PTOperator.State.PENDING_DEPLOY, o1p1.getState());
 
@@ -587,11 +596,10 @@ public class StreamingContainerManagerTest
 
   }
 
-  public static class TestStaticPartitioningSerDe extends DefaultStatefulStreamCodec<Object> {
+  public static class TestStaticPartitioningSerDe extends DefaultStatefulStreamCodec<Object>
+  {
 
-    public final static int[] partitions = new int[]{
-      0, 1, 2
-    };
+    public static final int[] partitions = new int[]{0, 1, 2};
 
     @Override
     public int getPartition(Object o)
@@ -604,7 +612,8 @@ public class StreamingContainerManagerTest
 
   }
 
-  private int countState(PTContainer c, PTOperator.State state) {
+  private int countState(PTContainer c, PTOperator.State state)
+  {
     int count = 0;
     for (PTOperator o : c.getOperators()) {
       if (o.getState() == state) {
@@ -614,15 +623,18 @@ public class StreamingContainerManagerTest
     return count;
   }
 
-  private boolean containsNodeContext(List<OperatorDeployInfo> di, OperatorMeta nodeConf) {
+  private boolean containsNodeContext(List<OperatorDeployInfo> di, OperatorMeta nodeConf)
+  {
     return getNodeDeployInfo(di, nodeConf) != null;
   }
 
-  public static List<OperatorDeployInfo> getDeployInfo(StreamingContainerAgent sca) {
+  public static List<OperatorDeployInfo> getDeployInfo(StreamingContainerAgent sca)
+  {
     return sca.getDeployInfoList(sca.container.getOperators());
   }
 
-  private static OperatorDeployInfo getNodeDeployInfo(List<OperatorDeployInfo> di, OperatorMeta nodeConf) {
+  private static OperatorDeployInfo getNodeDeployInfo(List<OperatorDeployInfo> di, OperatorMeta nodeConf)
+  {
     for (OperatorDeployInfo ndi : di) {
       if (nodeConf.getName().equals(ndi.name)) {
         return ndi;
@@ -631,7 +643,8 @@ public class StreamingContainerManagerTest
     return null;
   }
 
-  private static InputDeployInfo getInputDeployInfo(OperatorDeployInfo ndi, String streamId) {
+  private static InputDeployInfo getInputDeployInfo(OperatorDeployInfo ndi, String streamId)
+  {
     for (InputDeployInfo in : ndi.inputs) {
       if (streamId.equals(in.declaredStreamId)) {
         return in;
@@ -640,8 +653,9 @@ public class StreamingContainerManagerTest
     return null;
   }
 
-  public static StreamingContainerAgent assignContainer(StreamingContainerManager scm, String containerId) {
-    return scm.assignContainer(new ContainerResource(0, containerId, "localhost", 1024, 0,null), InetSocketAddress.createUnresolved(containerId+"Host", 0));
+  public static StreamingContainerAgent assignContainer(StreamingContainerManager scm, String containerId)
+  {
+    return scm.assignContainer(new ContainerResource(0, containerId, "localhost", 1024, 0, null), InetSocketAddress.createUnresolved(containerId + "Host", 0));
   }
 
   @Test
@@ -846,7 +860,7 @@ public class StreamingContainerManagerTest
   }
 
   private void setupAppDataSourceLogicalPlan(Class<? extends TestAppDataQueryOperator> qClass,
-          Class<? extends TestAppDataSourceOperator> dsClass, Class<? extends TestAppDataResultOperator> rClass)
+      Class<? extends TestAppDataSourceOperator> dsClass, Class<? extends TestAppDataResultOperator> rClass)
   {
     TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class);
     TestAppDataQueryOperator q = dag.addOperator("q", qClass);
@@ -1086,9 +1100,7 @@ public class StreamingContainerManagerTest
     Assert.assertTrue("Latency must be greater than or equal to zero", o4Info.latencyMA >= 0);
     StreamingContainerManager.CriticalPathInfo criticalPathInfo = dnmgr.getCriticalPathInfo();
     Assert.assertArrayEquals("Critical Path must be the path in the DAG that includes the HighLatencyTestOperator",
-        new Integer[]{o1Info.partitions.iterator().next(),
-            o3Info.partitions.iterator().next(),
-            o4Info.partitions.iterator().next()},
+        new Integer[]{o1Info.partitions.iterator().next(), o3Info.partitions.iterator().next(), o4Info.partitions.iterator().next()},
         criticalPathInfo.path.toArray());
     Assert.assertTrue("Whole DAG latency must be greater than the artificially introduced latency",
         criticalPathInfo.latency > latency);

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java b/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
index 59a45aa..ae1bcb4 100644
--- a/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/cli/DTCliTest.java
@@ -19,22 +19,24 @@
 package com.datatorrent.stram.cli;
 
 import java.io.File;
-import com.datatorrent.stram.client.AppPackage;
-import com.datatorrent.stram.client.ConfigPackage;
-import com.datatorrent.stram.client.DTConfiguration;
-import com.datatorrent.stram.support.StramTestSupport;
-import com.datatorrent.stram.support.StramTestSupport.TestHomeDirectory;
-import static com.datatorrent.stram.support.StramTestSupport.setEnv;
 import java.util.HashMap;
 import java.util.Map;
-import org.apache.commons.io.FileUtils;
+
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
+import org.apache.commons.io.FileUtils;
+
+import com.datatorrent.stram.client.AppPackage;
+import com.datatorrent.stram.client.ConfigPackage;
+import com.datatorrent.stram.client.DTConfiguration;
+import com.datatorrent.stram.support.StramTestSupport;
+
+import static com.datatorrent.stram.support.StramTestSupport.setEnv;
+
 /**
  *
  */

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java b/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
index 5400930..5f16399 100644
--- a/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/client/AppPackageTest.java
@@ -18,22 +18,24 @@
  */
 package com.datatorrent.stram.client;
 
-import com.datatorrent.stram.support.StramTestSupport;
-import com.datatorrent.stram.util.JSONSerializationProvider;
-import net.lingala.zip4j.exception.ZipException;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.IOUtils;
-import org.codehaus.jettison.json.JSONException;
-import org.codehaus.jettison.json.JSONObject;
-import org.junit.Assert;
-import org.junit.Test;
-
 import java.io.File;
 import java.io.IOException;
 import java.util.List;
 import java.util.Set;
+
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.commons.io.IOUtils;
+
+import com.datatorrent.stram.support.StramTestSupport;
+import com.datatorrent.stram.util.JSONSerializationProvider;
+
+import net.lingala.zip4j.exception.ZipException;
 
 /**
  *

http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/85db448e/engine/src/test/java/com/datatorrent/stram/client/ConfigPackageTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/client/ConfigPackageTest.java b/engine/src/test/java/com/datatorrent/stram/client/ConfigPackageTest.java
index d02f3fc..88545cb 100644
--- a/engine/src/test/java/com/datatorrent/stram/client/ConfigPackageTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/client/ConfigPackageTest.java
@@ -18,10 +18,10 @@
  */
 package com.datatorrent.stram.client;
 
-import com.datatorrent.stram.support.StramTestSupport;
-import com.datatorrent.stram.util.JSONSerializationProvider;
-import net.lingala.zip4j.exception.ZipException;
-import org.apache.commons.io.FileUtils;
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
 import org.junit.Assert;
@@ -31,9 +31,12 @@ import org.junit.rules.TemporaryFolder;
 import org.junit.rules.TestWatcher;
 import org.junit.runner.Description;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.Map;
+import org.apache.commons.io.FileUtils;
+
+import com.datatorrent.stram.support.StramTestSupport;
+import com.datatorrent.stram.util.JSONSerializationProvider;
+
+import net.lingala.zip4j.exception.ZipException;
 
 /**
  *



Mime
View raw message