engine/src/test/java/com/datatorrent/stram/support/StramTestSupport.java - apex-core - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package com.datatorrent.stram.support;

 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
 import java.lang.reflect.Field;
 import java.net.URI;
 import java.nio.file.FileAlreadyExistsException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;

 import javax.servlet.http.HttpServletRequest;

 import org.codehaus.plexus.DefaultPlexusContainer;
 import org.codehaus.plexus.PlexusContainer;
 import org.codehaus.plexus.logging.BaseLoggerManager;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.nio.SelectChannelConnector;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
 import org.eclipse.jetty.websocket.WebSocket;
 import org.eclipse.jetty.websocket.WebSocketServlet;
 import org.hamcrest.BaseMatcher;
 import org.hamcrest.Description;
 import org.junit.Assert;
 import org.junit.rules.TestWatcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.util.Clock;
 import org.apache.maven.cli.MavenCli;
 import org.apache.maven.cli.logging.Slf4jLogger;

 import com.datatorrent.api.Attribute;
 import com.datatorrent.api.StorageAgent;
 import com.datatorrent.bufferserver.packet.MessageType;
 import com.datatorrent.stram.StramAppContext;
 import com.datatorrent.stram.StramLocalCluster;
 import com.datatorrent.stram.StramLocalCluster.LocalStreamingContainer;
 import com.datatorrent.stram.api.AppDataSource;
 import com.datatorrent.stram.api.BaseContext;
 import com.datatorrent.stram.engine.OperatorContext;
 import com.datatorrent.stram.engine.WindowGenerator;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.plan.physical.PTOperator;
 import com.datatorrent.stram.tuple.EndWindowTuple;
 import com.datatorrent.stram.tuple.Tuple;
 import com.datatorrent.stram.webapp.AppInfo;

 import net.lingala.zip4j.core.ZipFile;
 import net.lingala.zip4j.model.ZipParameters;

 /**
  * Bunch of utilities shared between tests.
  */
 public abstract class StramTestSupport
 {
   private static final Logger LOG = LoggerFactory.getLogger(StramTestSupport.class);
   private static MavenCli mavenCli = new MavenCli()
   {
     @Override
     protected void customizeContainer(PlexusContainer container)
     {
       ((DefaultPlexusContainer)container).setLoggerManager(
           new BaseLoggerManager()
           {
             @Override
             protected org.codehaus.plexus.logging.Logger createLogger(String s)
             {
               return new Slf4jLogger(LOG);
             }
           }
       );
     }
   };
   private static final String workingDirectory = "src/test/resources/testAppPackage/mydtapp/";
   public static final long DEFAULT_TIMEOUT_MILLIS = 30000;

   public static Object generateTuple(Object payload, int windowId)
   {
     return payload;
   }

   public static Tuple generateBeginWindowTuple(String nodeid, int windowId)
   {
     Tuple bwt = new Tuple(MessageType.BEGIN_WINDOW, windowId);
     return bwt;
   }

   public static Tuple generateEndWindowTuple(String nodeid, int windowId)
   {
     EndWindowTuple t = new EndWindowTuple(windowId);
     return t;
   }

   public static void checkStringMatch(String print, String expected, String got)
   {
     Assert.assertTrue(print + " doesn't match, got: " + got + " expected: " + expected,
         got.matches(expected));
   }

   public static WindowGenerator setupWindowGenerator(ManualScheduledExecutorService mses)
   {
     WindowGenerator gen = new WindowGenerator(mses, 1024);
     gen.setResetWindow(0);
     gen.setFirstWindow(0);
     gen.setWindowWidth(1);
     return gen;
   }

   @SuppressWarnings("SleepWhileInLoop")
   public static void waitForWindowComplete(OperatorContext nodeCtx, long windowId) throws InterruptedException
   {
     LOG.debug("Waiting for end of window {} at node {} when lastProcessedWindowId is {}", new Object[] {windowId, nodeCtx.getId(), nodeCtx.getLastProcessedWindowId()});
     long startMillis = System.currentTimeMillis();
     while (nodeCtx.getLastProcessedWindowId() < windowId) {
       if (System.currentTimeMillis() > (startMillis + DEFAULT_TIMEOUT_MILLIS)) {
         long timeout = System.currentTimeMillis() - startMillis;
         throw new AssertionError(String.format("Timeout %s ms waiting for window %s operator %s", timeout, windowId, nodeCtx.getId()));
       }
       Thread.sleep(20);
     }
   }

   /**
    * Create an appPackage zip using the sample appPackage located in
    * src/test/resources/testAppPackage/testAppPackageSrc.
    * @return      The File object that can be used in the AppPackage constructor.
    */
   public static File createAppPackageFile()
   {
     final String version = System.getProperty("apex.version");
     final List<String> params = new LinkedList<>();
     params.add("clean");
     params.add("package");
     params.add("-DskipTests");
     if (version != null && version.length() > 0) {
       params.add("-Dapex.version=" + version);
     }
     Assert.assertEquals(0, mavenCli.doMain(params.toArray(new String[params.size()]), workingDirectory, System.out, System.err));
     return new File(workingDirectory, "target/mydtapp-1.0-SNAPSHOT.apa");
   }

   public static void removeAppPackageFile()
   {
     Assert.assertEquals(0, mavenCli.doMain(new String[]{"clean"}, workingDirectory, System.out, System.err));
   }

   /**
    * Create an confPackage zip using the sample confPackage located in
    * src/test/resources/testConfPackage/testConfPackageSrc.
    *
    * @param file The file whose path will be used to create the confPackage zip
    * @return The File object that can be used in the ConfigPackage constructor.
    * @throws net.lingala.zip4j.exception.ZipException
    */
   public static File createConfigPackageFile(File file) throws net.lingala.zip4j.exception.ZipException
   {
     ZipFile zipFile = new ZipFile(file);
     ZipParameters zipParameters = new ZipParameters();
     zipParameters.setIncludeRootFolder(false);
     zipFile.createZipFileFromFolder("src/test/resources/testConfigPackage/testConfigPackageSrc", zipParameters, false, Long.MAX_VALUE);
     return file;
   }

   public interface WaitCondition
   {
     boolean isComplete();

   }

   @SuppressWarnings("SleepWhileInLoop")
   public static boolean awaitCompletion(WaitCondition c, long timeoutMillis) throws InterruptedException
   {
     long startMillis = System.currentTimeMillis();
     while (System.currentTimeMillis() < (startMillis + timeoutMillis)) {
       if (c.isComplete()) {
         return true;
       }
       Thread.sleep(50);
     }
     return c.isComplete();
   }

   /**
    * Wait until instance of operator is deployed into a container and return the container reference.
    * Asserts non null return value.
    *
    * @param localCluster
    * @param operator
    * @return
    * @throws InterruptedException
    */
   @SuppressWarnings("SleepWhileInLoop")
   public static LocalStreamingContainer waitForActivation(StramLocalCluster localCluster, PTOperator operator) throws InterruptedException
   {
     LocalStreamingContainer container;
     long startMillis = System.currentTimeMillis();
     while (System.currentTimeMillis() < (startMillis + DEFAULT_TIMEOUT_MILLIS)) {
       if (operator.getState() == PTOperator.State.ACTIVE) {
         if ((container = localCluster.getContainer(operator)) != null) {
           return container;
         }
       }
       LOG.debug("Waiting for {}({}) in container {}", new Object[] {operator, operator.getState(), operator.getContainer()});
       Thread.sleep(500);
     }

     Assert.fail("timeout waiting for operator deployment " + operator);
     return null;
   }

   public static class RegexMatcher extends BaseMatcher<String>
   {
     private final String regex;

     public RegexMatcher(String regex)
     {
       this.regex = regex;
     }

     @Override
     public boolean matches(Object o)
     {
       return ((String)o).matches(regex);

     }

     @Override
     public void describeTo(Description description)
     {
       description.appendText("matches regex=" + regex);
     }

     public static RegexMatcher matches(String regex)
     {
       return new RegexMatcher(regex);
     }

   }

   public static class TestMeta extends TestWatcher
   {
     private File dir;

     @Override
     protected void starting(org.junit.runner.Description description)
     {
       final String methodName = description.getMethodName();
       final String className = description.getClassName();
       dir = new File("target/" + className + "/" + methodName);
       try {
         Files.createDirectories(dir.toPath());
       } catch (FileAlreadyExistsException e) {
         try {
           Files.delete(dir.toPath());
           Files.createDirectories(dir.toPath());
         } catch (IOException ioe) {
           throw new RuntimeException("Fail to create test working directory " + dir.getAbsolutePath(), e);
         }
       } catch (IOException e) {
         throw new RuntimeException("Fail to create test working directory " + dir.getAbsolutePath(), e);
       }
     }

     @Override
     protected void finished(org.junit.runner.Description description)
     {
       FileUtils.deleteQuietly(dir);
     }

     public String getPath()
     {
       return dir.getPath();
     }

     public String getAbsolutePath()
     {
       return dir.getAbsolutePath();
     }

     public Path toPath()
     {
       return dir.toPath();
     }

     public URI toURI()
     {
       return dir.toURI();
     }

   }

   public static class TestHomeDirectory extends TestWatcher
   {

     Map<String, String> env = new HashMap<>();
     String userHome;

     @Override
     protected void starting(org.junit.runner.Description description)
     {
       super.starting(description);
       try {
         userHome = System.getProperty("user.home");
         env.put("HOME", System.getProperty("user.dir") + "/src/test/resources/testAppPackage");
         setEnv(env);
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
     }

     @Override
     protected void finished(org.junit.runner.Description description)
     {
       super.finished(description);

       try {
         env.put("HOME", userHome);
         setEnv(env);
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
     }
   }

   public static void setEnv(Map<String, String> newenv) throws Exception
   {
     try {
       Class<?> processEnvironmentClass = Class.forName("java.lang.ProcessEnvironment");
       Field theEnvironmentField = processEnvironmentClass.getDeclaredField("theEnvironment");
       theEnvironmentField.setAccessible(true);
       Map<String, String> env = (Map<String, String>)theEnvironmentField.get(null);
       env.putAll(newenv);
       Field theCaseInsensitiveEnvironmentField = processEnvironmentClass.getDeclaredField("theCaseInsensitiveEnvironment");
       theCaseInsensitiveEnvironmentField.setAccessible(true);
       Map<String, String> cienv = (Map<String, String>)theCaseInsensitiveEnvironmentField.get(null);
       cienv.putAll(newenv);
     } catch (NoSuchFieldException e) {
       Class[] classes = Collections.class.getDeclaredClasses();
       Map<String, String> env = System.getenv();
       for (Class cl : classes) {
         if ("java.util.Collections$UnmodifiableMap".equals(cl.getName())) {
           Field field = cl.getDeclaredField("m");
           field.setAccessible(true);
           Object obj = field.get(env);
           Map<String, String> map = (Map<String, String>)obj;
           map.clear();
           map.putAll(newenv);
         }
       }
     }
   }

   public static LogicalPlan createDAG(final TestMeta testMeta, final String suffix)
   {
     if (suffix == null) {
       throw new NullPointerException();
     }
     LogicalPlan dag = new LogicalPlan();
     dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.getPath() + suffix);
     return dag;
   }

   public static LogicalPlan createDAG(final TestMeta testMeta)
   {
     return createDAG(testMeta, "");
   }

   public static boolean isInTravis()
   {
     return "true".equals(System.getProperty("travis"));
   }

   public static class MemoryStorageAgent implements StorageAgent, Serializable
   {
     static class OperatorWindowIdPair implements Serializable
     {
       final int operatorId;
       final long windowId;

       OperatorWindowIdPair(int operatorId, long windowId)
       {
         this.operatorId = operatorId;
         this.windowId = windowId;
       }

       @Override
       public int hashCode()
       {
         int hash = 7;
         hash = 97 * hash + this.operatorId;
         hash = 97 * hash + (int)(this.windowId ^ (this.windowId >>> 32));
         return hash;
       }

       @Override
       public boolean equals(Object obj)
       {
         if (obj == null) {
           return false;
         }
         if (getClass() != obj.getClass()) {
           return false;
         }
         final OperatorWindowIdPair other = (OperatorWindowIdPair)obj;
         if (this.operatorId != other.operatorId) {
           return false;
         }
         if (this.windowId != other.windowId) {
           return false;
         }
         return true;
       }

       private static final long serialVersionUID = 201404091805L;
     }

     transient HashMap<OperatorWindowIdPair, Object> store = new HashMap<>();

     @Override
     public synchronized void save(Object object, int operatorId, long windowId) throws IOException
     {
       store.put(new OperatorWindowIdPair(operatorId, windowId), object);
     }

     @Override
     public synchronized Object load(int operatorId, long windowId) throws IOException
     {
       return store.get(new OperatorWindowIdPair(operatorId, windowId));
     }

     @Override
     public synchronized void delete(int operatorId, long windowId) throws IOException
     {
       store.remove(new OperatorWindowIdPair(operatorId, windowId));
     }

     @Override
     public synchronized long[] getWindowIds(int operatorId) throws IOException
     {
       ArrayList<Long> windowIds = new ArrayList<>();
       for (OperatorWindowIdPair key : store.keySet()) {
         if (key.operatorId == operatorId) {
           windowIds.add(key.windowId);
         }
       }

       long[] ret = new long[windowIds.size()];
       for (int i = ret.length; i-- > 0;) {
         ret[i] = windowIds.get(i).longValue();
       }

       return ret;
     }

     private static final long serialVersionUID = 201404091747L;
   }

   public static class TestAppContext extends BaseContext implements StramAppContext
   {

     final ApplicationAttemptId appAttemptID;
     final ApplicationId appID;
     final String appPath = "/testPath";
     final String userId = "testUser";
     final long startTime = System.currentTimeMillis();
     final String gatewayAddress = "localhost:9090";

     public TestAppContext(Attribute.AttributeMap attributeMap, int appid, int numJobs, int numTasks, int numAttempts)
     {
       super(attributeMap, null); // this needs to be done in a proper way - may cause application errors.
       this.appID = ApplicationId.newInstance(0, appid);
       this.appAttemptID = ApplicationAttemptId.newInstance(this.appID, numAttempts);
     }

     public TestAppContext(Attribute.AttributeMap attributeMap)
     {
       this(attributeMap, 0, 1, 1, 1);
     }

     @Override
     public ApplicationAttemptId getApplicationAttemptId()
     {
       return appAttemptID;
     }

     @Override
     public ApplicationId getApplicationID()
     {
       return appID;
     }

     @Override
     public String getApplicationPath()
     {
       return appPath;
     }

     @Override
     public String getAppMasterTrackingUrl()
     {
       return "unknown";
     }

     @Override
     public CharSequence getUser()
     {
       return userId;
     }

     @Override
     public Clock getClock()
     {
       return null;
     }

     @Override
     public String getApplicationName()
     {
       return "TestApp";
     }

     @Override
     public String getApplicationDocLink()
     {
       return "TestAppDocLink";
     }

     @Override
     public long getStartTime()
     {
       return startTime;
     }

     @Override
     public AppInfo.AppStats getStats()
     {
       return new AppInfo.AppStats()
       {
       };
     }

     @Override
     public String getGatewayAddress()
     {
       return gatewayAddress;
     }

     @Override
     public boolean isGatewayConnected()
     {
       return false;
     }

     @Override
     public List<AppDataSource> getAppDataSources()
     {
       return null;
     }

     @Override
     public Map<String, Object> getMetrics()
     {
       return null;
     }

     @SuppressWarnings("FieldNameHidesFieldInSuperclass")
     private static final long serialVersionUID = 201309121323L;
   }

   public static class EmbeddedWebSocketServer
   {

     private final Logger LOG = LoggerFactory.getLogger(EmbeddedWebSocketServer.class);

     private int port;
     private Server server;
     private WebSocket websocket;

     public EmbeddedWebSocketServer(int port)
     {
       this.port = port;
     }

     public void setWebSocket(WebSocket websocket)
     {
       this.websocket = websocket;
     }

     public void start() throws Exception
     {
       server = new Server();
       Connector connector = new SelectChannelConnector();
       connector.setPort(port);
       server.addConnector(connector);

         // Setup the basic application "context" for this application at "/"
       // This is also known as the handler tree (in jetty speak)
       ServletContextHandler contextHandler = new ServletContextHandler(ServletContextHandler.SESSIONS);
       contextHandler.setContextPath("/");
       server.setHandler(contextHandler);
       WebSocketServlet webSocketServlet = new WebSocketServlet()
       {
         @Override
         public WebSocket doWebSocketConnect(HttpServletRequest request, String protocol)
         {
           return websocket;
         }
       };

       contextHandler.addServlet(new ServletHolder(webSocketServlet), "/pubsub");
       server.start();
       if (port == 0) {
         port = server.getConnectors()[0].getLocalPort();
       }
     }

     public int getPort()
     {
       return port;
     }

     public void stop() throws Exception
     {
       server.stop();
     }
   }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package com.datatorrent.stram.support;

	import java.io.File;
	import java.io.IOException;
	import java.io.Serializable;
	import java.lang.reflect.Field;
	import java.net.URI;
	import java.nio.file.FileAlreadyExistsException;
	import java.nio.file.Files;
	import java.nio.file.Path;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Map;

	import javax.servlet.http.HttpServletRequest;

	import org.codehaus.plexus.DefaultPlexusContainer;
	import org.codehaus.plexus.PlexusContainer;
	import org.codehaus.plexus.logging.BaseLoggerManager;
	import org.eclipse.jetty.server.Connector;
	import org.eclipse.jetty.server.Server;
	import org.eclipse.jetty.server.nio.SelectChannelConnector;
	import org.eclipse.jetty.servlet.ServletContextHandler;
	import org.eclipse.jetty.servlet.ServletHolder;
	import org.eclipse.jetty.websocket.WebSocket;
	import org.eclipse.jetty.websocket.WebSocketServlet;
	import org.hamcrest.BaseMatcher;
	import org.hamcrest.Description;
	import org.junit.Assert;
	import org.junit.rules.TestWatcher;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.commons.io.FileUtils;
	import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
	import org.apache.hadoop.yarn.api.records.ApplicationId;
	import org.apache.hadoop.yarn.util.Clock;
	import org.apache.maven.cli.MavenCli;
	import org.apache.maven.cli.logging.Slf4jLogger;

	import com.datatorrent.api.Attribute;
	import com.datatorrent.api.StorageAgent;
	import com.datatorrent.bufferserver.packet.MessageType;
	import com.datatorrent.stram.StramAppContext;
	import com.datatorrent.stram.StramLocalCluster;
	import com.datatorrent.stram.StramLocalCluster.LocalStreamingContainer;
	import com.datatorrent.stram.api.AppDataSource;
	import com.datatorrent.stram.api.BaseContext;
	import com.datatorrent.stram.engine.OperatorContext;
	import com.datatorrent.stram.engine.WindowGenerator;
	import com.datatorrent.stram.plan.logical.LogicalPlan;
	import com.datatorrent.stram.plan.physical.PTOperator;
	import com.datatorrent.stram.tuple.EndWindowTuple;
	import com.datatorrent.stram.tuple.Tuple;
	import com.datatorrent.stram.webapp.AppInfo;

	import net.lingala.zip4j.core.ZipFile;
	import net.lingala.zip4j.model.ZipParameters;

	/**
	* Bunch of utilities shared between tests.
	*/
	public abstract class StramTestSupport
	{
	private static final Logger LOG = LoggerFactory.getLogger(StramTestSupport.class);
	private static MavenCli mavenCli = new MavenCli()
	{
	@Override
	protected void customizeContainer(PlexusContainer container)
	{
	((DefaultPlexusContainer)container).setLoggerManager(
	new BaseLoggerManager()
	{
	@Override
	protected org.codehaus.plexus.logging.Logger createLogger(String s)
	{
	return new Slf4jLogger(LOG);
	}
	}
	);
	}
	};
	private static final String workingDirectory = "src/test/resources/testAppPackage/mydtapp/";
	public static final long DEFAULT_TIMEOUT_MILLIS = 30000;

	public static Object generateTuple(Object payload, int windowId)
	{
	return payload;
	}

	public static Tuple generateBeginWindowTuple(String nodeid, int windowId)
	{
	Tuple bwt = new Tuple(MessageType.BEGIN_WINDOW, windowId);
	return bwt;
	}

	public static Tuple generateEndWindowTuple(String nodeid, int windowId)
	{
	EndWindowTuple t = new EndWindowTuple(windowId);
	return t;
	}

	public static void checkStringMatch(String print, String expected, String got)
	{
	Assert.assertTrue(print + " doesn't match, got: " + got + " expected: " + expected,
	got.matches(expected));
	}

	public static WindowGenerator setupWindowGenerator(ManualScheduledExecutorService mses)
	{
	WindowGenerator gen = new WindowGenerator(mses, 1024);
	gen.setResetWindow(0);
	gen.setFirstWindow(0);
	gen.setWindowWidth(1);
	return gen;
	}

	@SuppressWarnings("SleepWhileInLoop")
	public static void waitForWindowComplete(OperatorContext nodeCtx, long windowId) throws InterruptedException
	{
	LOG.debug("Waiting for end of window {} at node {} when lastProcessedWindowId is {}", new Object[] {windowId, nodeCtx.getId(), nodeCtx.getLastProcessedWindowId()});
	long startMillis = System.currentTimeMillis();
	while (nodeCtx.getLastProcessedWindowId() < windowId) {
	if (System.currentTimeMillis() > (startMillis + DEFAULT_TIMEOUT_MILLIS)) {
	long timeout = System.currentTimeMillis() - startMillis;
	throw new AssertionError(String.format("Timeout %s ms waiting for window %s operator %s", timeout, windowId, nodeCtx.getId()));
	}
	Thread.sleep(20);
	}
	}

	/**
	* Create an appPackage zip using the sample appPackage located in
	* src/test/resources/testAppPackage/testAppPackageSrc.
	* @return The File object that can be used in the AppPackage constructor.
	*/
	public static File createAppPackageFile()
	{
	final String version = System.getProperty("apex.version");
	final List<String> params = new LinkedList<>();
	params.add("clean");
	params.add("package");
	params.add("-DskipTests");
	if (version != null && version.length() > 0) {
	params.add("-Dapex.version=" + version);
	}
	Assert.assertEquals(0, mavenCli.doMain(params.toArray(new String[params.size()]), workingDirectory, System.out, System.err));
	return new File(workingDirectory, "target/mydtapp-1.0-SNAPSHOT.apa");
	}

	public static void removeAppPackageFile()
	{
	Assert.assertEquals(0, mavenCli.doMain(new String[]{"clean"}, workingDirectory, System.out, System.err));
	}

	/**
	* Create an confPackage zip using the sample confPackage located in
	* src/test/resources/testConfPackage/testConfPackageSrc.
	*
	* @param file The file whose path will be used to create the confPackage zip
	* @return The File object that can be used in the ConfigPackage constructor.
	* @throws net.lingala.zip4j.exception.ZipException
	*/
	public static File createConfigPackageFile(File file) throws net.lingala.zip4j.exception.ZipException
	{
	ZipFile zipFile = new ZipFile(file);
	ZipParameters zipParameters = new ZipParameters();
	zipParameters.setIncludeRootFolder(false);
	zipFile.createZipFileFromFolder("src/test/resources/testConfigPackage/testConfigPackageSrc", zipParameters, false, Long.MAX_VALUE);
	return file;
	}

	public interface WaitCondition
	{
	boolean isComplete();

	}

	@SuppressWarnings("SleepWhileInLoop")
	public static boolean awaitCompletion(WaitCondition c, long timeoutMillis) throws InterruptedException
	{
	long startMillis = System.currentTimeMillis();
	while (System.currentTimeMillis() < (startMillis + timeoutMillis)) {
	if (c.isComplete()) {
	return true;
	}
	Thread.sleep(50);
	}
	return c.isComplete();
	}

	/**
	* Wait until instance of operator is deployed into a container and return the container reference.
	* Asserts non null return value.
	*
	* @param localCluster
	* @param operator
	* @return
	* @throws InterruptedException
	*/
	@SuppressWarnings("SleepWhileInLoop")
	public static LocalStreamingContainer waitForActivation(StramLocalCluster localCluster, PTOperator operator) throws InterruptedException
	{
	LocalStreamingContainer container;
	long startMillis = System.currentTimeMillis();
	while (System.currentTimeMillis() < (startMillis + DEFAULT_TIMEOUT_MILLIS)) {
	if (operator.getState() == PTOperator.State.ACTIVE) {
	if ((container = localCluster.getContainer(operator)) != null) {
	return container;
	}
	}
	LOG.debug("Waiting for {}({}) in container {}", new Object[] {operator, operator.getState(), operator.getContainer()});
	Thread.sleep(500);
	}

	Assert.fail("timeout waiting for operator deployment " + operator);
	return null;
	}

	public static class RegexMatcher extends BaseMatcher<String>
	{
	private final String regex;

	public RegexMatcher(String regex)
	{
	this.regex = regex;
	}

	@Override
	public boolean matches(Object o)
	{
	return ((String)o).matches(regex);

	}

	@Override
	public void describeTo(Description description)
	{
	description.appendText("matches regex=" + regex);
	}

	public static RegexMatcher matches(String regex)
	{
	return new RegexMatcher(regex);
	}

	}

	public static class TestMeta extends TestWatcher
	{
	private File dir;

	@Override
	protected void starting(org.junit.runner.Description description)
	{
	final String methodName = description.getMethodName();
	final String className = description.getClassName();
	dir = new File("target/" + className + "/" + methodName);
	try {
	Files.createDirectories(dir.toPath());
	} catch (FileAlreadyExistsException e) {
	try {
	Files.delete(dir.toPath());
	Files.createDirectories(dir.toPath());
	} catch (IOException ioe) {
	throw new RuntimeException("Fail to create test working directory " + dir.getAbsolutePath(), e);
	}
	} catch (IOException e) {
	throw new RuntimeException("Fail to create test working directory " + dir.getAbsolutePath(), e);
	}
	}

	@Override
	protected void finished(org.junit.runner.Description description)
	{
	FileUtils.deleteQuietly(dir);
	}

	public String getPath()
	{
	return dir.getPath();
	}

	public String getAbsolutePath()
	{
	return dir.getAbsolutePath();
	}

	public Path toPath()
	{
	return dir.toPath();
	}

	public URI toURI()
	{
	return dir.toURI();
	}

	}

	public static class TestHomeDirectory extends TestWatcher
	{

	Map<String, String> env = new HashMap<>();
	String userHome;

	@Override
	protected void starting(org.junit.runner.Description description)
	{
	super.starting(description);
	try {
	userHome = System.getProperty("user.home");
	env.put("HOME", System.getProperty("user.dir") + "/src/test/resources/testAppPackage");
	setEnv(env);
	} catch (Exception e) {
	throw new RuntimeException(e);
	}
	}

	@Override
	protected void finished(org.junit.runner.Description description)
	{
	super.finished(description);

	try {
	env.put("HOME", userHome);
	setEnv(env);
	} catch (Exception e) {
	throw new RuntimeException(e);
	}
	}
	}

	public static void setEnv(Map<String, String> newenv) throws Exception
	{
	try {
	Class<?> processEnvironmentClass = Class.forName("java.lang.ProcessEnvironment");
	Field theEnvironmentField = processEnvironmentClass.getDeclaredField("theEnvironment");
	theEnvironmentField.setAccessible(true);
	Map<String, String> env = (Map<String, String>)theEnvironmentField.get(null);
	env.putAll(newenv);
	Field theCaseInsensitiveEnvironmentField = processEnvironmentClass.getDeclaredField("theCaseInsensitiveEnvironment");
	theCaseInsensitiveEnvironmentField.setAccessible(true);
	Map<String, String> cienv = (Map<String, String>)theCaseInsensitiveEnvironmentField.get(null);
	cienv.putAll(newenv);
	} catch (NoSuchFieldException e) {
	Class[] classes = Collections.class.getDeclaredClasses();
	Map<String, String> env = System.getenv();
	for (Class cl : classes) {
	if ("java.util.Collections$UnmodifiableMap".equals(cl.getName())) {
	Field field = cl.getDeclaredField("m");
	field.setAccessible(true);
	Object obj = field.get(env);
	Map<String, String> map = (Map<String, String>)obj;
	map.clear();
	map.putAll(newenv);
	}
	}
	}
	}

	public static LogicalPlan createDAG(final TestMeta testMeta, final String suffix)
	{
	if (suffix == null) {
	throw new NullPointerException();
	}
	LogicalPlan dag = new LogicalPlan();
	dag.setAttribute(LogicalPlan.APPLICATION_PATH, testMeta.getPath() + suffix);
	return dag;
	}

	public static LogicalPlan createDAG(final TestMeta testMeta)
	{
	return createDAG(testMeta, "");
	}

	public static boolean isInTravis()
	{
	return "true".equals(System.getProperty("travis"));
	}

	public static class MemoryStorageAgent implements StorageAgent, Serializable
	{
	static class OperatorWindowIdPair implements Serializable
	{
	final int operatorId;
	final long windowId;

	OperatorWindowIdPair(int operatorId, long windowId)
	{
	this.operatorId = operatorId;
	this.windowId = windowId;
	}

	@Override
	public int hashCode()
	{
	int hash = 7;
	hash = 97 * hash + this.operatorId;
	hash = 97 * hash + (int)(this.windowId ^ (this.windowId >>> 32));
	return hash;
	}

	@Override
	public boolean equals(Object obj)
	{
	if (obj == null) {
	return false;
	}
	if (getClass() != obj.getClass()) {
	return false;
	}
	final OperatorWindowIdPair other = (OperatorWindowIdPair)obj;
	if (this.operatorId != other.operatorId) {
	return false;
	}
	if (this.windowId != other.windowId) {
	return false;
	}
	return true;
	}

	private static final long serialVersionUID = 201404091805L;
	}

	transient HashMap<OperatorWindowIdPair, Object> store = new HashMap<>();

	@Override
	public synchronized void save(Object object, int operatorId, long windowId) throws IOException
	{
	store.put(new OperatorWindowIdPair(operatorId, windowId), object);
	}

	@Override
	public synchronized Object load(int operatorId, long windowId) throws IOException
	{
	return store.get(new OperatorWindowIdPair(operatorId, windowId));
	}

	@Override
	public synchronized void delete(int operatorId, long windowId) throws IOException
	{
	store.remove(new OperatorWindowIdPair(operatorId, windowId));
	}

	@Override
	public synchronized long[] getWindowIds(int operatorId) throws IOException
	{
	ArrayList<Long> windowIds = new ArrayList<>();
	for (OperatorWindowIdPair key : store.keySet()) {
	if (key.operatorId == operatorId) {
	windowIds.add(key.windowId);
	}
	}

	long[] ret = new long[windowIds.size()];
	for (int i = ret.length; i-- > 0;) {
	ret[i] = windowIds.get(i).longValue();
	}

	return ret;
	}

	private static final long serialVersionUID = 201404091747L;
	}

	public static class TestAppContext extends BaseContext implements StramAppContext
	{

	final ApplicationAttemptId appAttemptID;
	final ApplicationId appID;
	final String appPath = "/testPath";
	final String userId = "testUser";
	final long startTime = System.currentTimeMillis();
	final String gatewayAddress = "localhost:9090";

	public TestAppContext(Attribute.AttributeMap attributeMap, int appid, int numJobs, int numTasks, int numAttempts)
	{
	super(attributeMap, null); // this needs to be done in a proper way - may cause application errors.
	this.appID = ApplicationId.newInstance(0, appid);
	this.appAttemptID = ApplicationAttemptId.newInstance(this.appID, numAttempts);
	}

	public TestAppContext(Attribute.AttributeMap attributeMap)
	{
	this(attributeMap, 0, 1, 1, 1);
	}

	@Override
	public ApplicationAttemptId getApplicationAttemptId()
	{
	return appAttemptID;
	}

	@Override
	public ApplicationId getApplicationID()
	{
	return appID;
	}

	@Override
	public String getApplicationPath()
	{
	return appPath;
	}

	@Override
	public String getAppMasterTrackingUrl()
	{
	return "unknown";
	}

	@Override
	public CharSequence getUser()
	{
	return userId;
	}

	@Override
	public Clock getClock()
	{
	return null;
	}

	@Override
	public String getApplicationName()
	{
	return "TestApp";
	}

	@Override
	public String getApplicationDocLink()
	{
	return "TestAppDocLink";
	}

	@Override
	public long getStartTime()
	{
	return startTime;
	}

	@Override
	public AppInfo.AppStats getStats()
	{
	return new AppInfo.AppStats()
	{
	};
	}

	@Override
	public String getGatewayAddress()
	{
	return gatewayAddress;
	}

	@Override
	public boolean isGatewayConnected()
	{
	return false;
	}

	@Override
	public List<AppDataSource> getAppDataSources()
	{
	return null;
	}

	@Override
	public Map<String, Object> getMetrics()
	{
	return null;
	}

	@SuppressWarnings("FieldNameHidesFieldInSuperclass")
	private static final long serialVersionUID = 201309121323L;
	}

	public static class EmbeddedWebSocketServer
	{

	private final Logger LOG = LoggerFactory.getLogger(EmbeddedWebSocketServer.class);

	private int port;
	private Server server;
	private WebSocket websocket;

	public EmbeddedWebSocketServer(int port)
	{
	this.port = port;
	}

	public void setWebSocket(WebSocket websocket)
	{
	this.websocket = websocket;
	}

	public void start() throws Exception
	{
	server = new Server();
	Connector connector = new SelectChannelConnector();
	connector.setPort(port);
	server.addConnector(connector);

	// Setup the basic application "context" for this application at "/"
	// This is also known as the handler tree (in jetty speak)
	ServletContextHandler contextHandler = new ServletContextHandler(ServletContextHandler.SESSIONS);
	contextHandler.setContextPath("/");
	server.setHandler(contextHandler);
	WebSocketServlet webSocketServlet = new WebSocketServlet()
	{
	@Override
	public WebSocket doWebSocketConnect(HttpServletRequest request, String protocol)
	{
	return websocket;
	}
	};

	contextHandler.addServlet(new ServletHolder(webSocketServlet), "/pubsub");
	server.start();
	if (port == 0) {
	port = server.getConnectors()[0].getLocalPort();
	}
	}

	public int getPort()
	{
	return port;
	}

	public void stop() throws Exception
	{
	server.stop();
	}
	}

	}