StramLocalCluster implements Runnable, Controller { public boolean isFinished() { return appDone; } StramLocalCluster(LogicalPlan dag);  StramLocalCluster(LogicalPlan dag, MockComponentFactory mcf); static LogicalPlan cloneLogicalPlan(LogicalPlan lp); StreamingContainerManager getStreamingContainerManager(); DAG getDAG(); PTOperator findByLogicalNode(OperatorMeta logicalNode); LocalStreamingContainer getContainer(PTOperator planOperator); @Override void runAsync(); @Override void shutdown(); boolean isFinished(); @Override void setHeartbeatMonitoringEnabled(boolean enabled); void setPerContainerBufferServer(boolean perContainerBufferServer); void setExitCondition(Callable<Boolean> exitCondition); void run(Callable<Boolean> exitCondition); void run(Callable<Boolean> exitCondition, long runMillis); @Override void run(); @Override @SuppressWarnings({"SleepWhileInLoop", "ResultOfObjectAllocationIgnored"}) void run(long runMillis);  }
@Test public void testAppPath() throws Exception { TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class); o1.setMaxTuples(1); File relPath = new File(dag.getAttributes().get(DAGContext.APPLICATION_PATH)); String uriPath = relPath.toURI().toString(); dag.setAttribute(DAGContext.APPLICATION_PATH, uriPath); StramLocalCluster cluster = new StramLocalCluster(dag); Assert.assertFalse(cluster.isFinished()); Assert.assertTrue("app path exists", relPath.exists() && relPath.isDirectory()); File checkPointDir = new File(relPath, LogicalPlan.SUBDIR_CHECKPOINTS); Assert.assertTrue("checkpoint path exists", checkPointDir.exists() && checkPointDir.isDirectory()); }