azkaban-memoizeit

diff --git a/src/java/azkaban/execapp/FlowRunner.java b/src/java/azkaban/execapp/FlowRunner.java
index fb9caf6..d38aa47 100644
--- a/src/java/azkaban/execapp/FlowRunner.java
+++ b/src/java/azkaban/execapp/FlowRunner.java
@@ -392,6 +392,7 @@ public class FlowRunner extends EventHandler implements Runnable {
 				flow.setStatus(Status.FAILED);
 			case FAILED:
 			case KILLED:
+			case FAILED_SUCCEEDED:
 				logger.info("Flow is set to " + flow.getStatus().toString());
 				break;
 			default:
@@ -674,6 +675,7 @@ public class FlowRunner extends EventHandler implements Runnable {
 		case KILLED:
 		case SKIPPED:
 		case SUCCEEDED:
+		case FAILED_SUCCEEDED:
 		case QUEUED:
 		case RUNNING:
 			return null;
@@ -692,6 +694,7 @@ public class FlowRunner extends EventHandler implements Runnable {
 				shouldKill = true;
 			case SKIPPED:
 			case SUCCEEDED:
+			case FAILED_SUCCEEDED:
 				continue;
 			case RUNNING:
 			case QUEUED:

diff --git a/src/java/azkaban/execapp/JobRunner.java b/src/java/azkaban/execapp/JobRunner.java
index 9096b2f..a363284 100644
--- a/src/java/azkaban/execapp/JobRunner.java
+++ b/src/java/azkaban/execapp/JobRunner.java
@@ -281,7 +281,7 @@ public class JobRunner extends EventHandler implements Runnable {
 			}
 			else {
 				node.setStatus(Status.FAILED);
-				logError("Job run failed!");
+				logError("Job run failed preparing the job.");
 			}
 			
 			node.setEndTime(System.currentTimeMillis());
@@ -362,7 +362,6 @@ public class JobRunner extends EventHandler implements Runnable {
 				}
 			}
 			
-			//job = JobWrappingFactory.getJobWrappingFactory().buildJobExecutor(node.getJobId(), props, logger);
 			try {
 				job = jobtypeManager.buildJobExecutor(node.getJobId(), props, logger);
 			}
@@ -380,10 +379,17 @@ public class JobRunner extends EventHandler implements Runnable {
 			job.run();
 		} catch (Exception e) {
 			e.printStackTrace();
-
-			node.setStatus(Status.FAILED);
-			logError("Job run failed!");
-			logError(e.getMessage() + e.getCause());
+			
+			if (props.getBoolean("job.succeed.on.failure", false)) {
+				node.setStatus(Status.FAILED_SUCCEEDED);
+				logError("Job run failed, but will treat it like success.");
+				logError(e.getMessage() + e.getCause());
+			}
+			else {
+				node.setStatus(Status.FAILED);
+				logError("Job run failed!");
+				logError(e.getMessage() + e.getCause());
+			}
 			return;
 		}
 

diff --git a/src/java/azkaban/executor/Status.java b/src/java/azkaban/executor/Status.java
index 1f4ee39..d8215df 100644
--- a/src/java/azkaban/executor/Status.java
+++ b/src/java/azkaban/executor/Status.java
@@ -1,7 +1,7 @@
 package azkaban.executor;
 
 public enum Status {
-	READY(10), PREPARING(20), RUNNING(30), PAUSED(40), SUCCEEDED(50), KILLED(60), FAILED(70), FAILED_FINISHING(80), SKIPPED(90), DISABLED(100), QUEUED(110);
+	READY(10), PREPARING(20), RUNNING(30), PAUSED(40), SUCCEEDED(50), KILLED(60), FAILED(70), FAILED_FINISHING(80), SKIPPED(90), DISABLED(100), QUEUED(110), FAILED_SUCCEEDED(120);
 	
 	private int numVal;
 
@@ -37,6 +37,8 @@ public enum Status {
 			return DISABLED;
 		case 110:
 			return QUEUED;
+		case 120:
+			return FAILED_SUCCEEDED;
 		default:
 			return READY;
 		}
@@ -48,6 +50,7 @@ public enum Status {
 		case KILLED:
 		case SUCCEEDED:
 		case SKIPPED:
+		case FAILED_SUCCEEDED:
 			return true;
 		default:
 			return false;

diff --git a/src/java/azkaban/flow/CommonJobProperties.java b/src/java/azkaban/flow/CommonJobProperties.java
index 35f308c..7a8ffc4 100644
--- a/src/java/azkaban/flow/CommonJobProperties.java
+++ b/src/java/azkaban/flow/CommonJobProperties.java
@@ -13,6 +13,11 @@ public class CommonJobProperties {
 	public static final String JOB_TYPE = "type";
 	
 	/**
+	 * Force a node to be a root node in a flow, even if there are other jobs dependent on it.
+	 */
+	public static final String ROOT_NODE = "root.node";
+	
+	/**
 	 * Comma delimited list of job names which are dependencies
 	 */
 	public static final String DEPENDENCIES = "dependencies";
@@ -95,5 +100,4 @@ public class CommonJobProperties {
 	public static final String FLOW_START_SECOND = "azkaban.flow.start.second";
 	public static final String FLOW_START_MILLISSECOND = "azkaban.flow.start.milliseconds";
 	public static final String FLOW_START_TIMEZONE = "azkaban.flow.start.timezone";
-
 }

diff --git a/src/java/azkaban/flow/Flow.java b/src/java/azkaban/flow/Flow.java
index 6c01df8..67375cf 100644
--- a/src/java/azkaban/flow/Flow.java
+++ b/src/java/azkaban/flow/Flow.java
@@ -37,7 +37,7 @@ public class Flow {
 	private HashMap<String, Set<Edge>> outEdges = new HashMap<String, Set<Edge>>();
 	private HashMap<String, Set<Edge>> inEdges = new HashMap<String, Set<Edge>>();
 	private HashMap<String, FlowProps> flowProps = new HashMap<String, FlowProps>(); 
-
+	
 	private List<String> failureEmail = new ArrayList<String>();
 	private List<String> successEmail = new ArrayList<String>();
 	private ArrayList<String> errors;
@@ -80,7 +80,7 @@ public class Flow {
 			}
 		}
 	}
-
+	
 	private void recursiveSetLevels(Node node) {
 		Set<Edge> edges = outEdges.get(node.getId());
 		if (edges != null) {
@@ -147,13 +147,13 @@ public class Flow {
 	public void addNode(Node node) {
 		nodes.put(node.getId(), node);
 	}
-
+	
 	public void addAllFlowProperties(Collection<FlowProps> props) {
 		for (FlowProps prop : props) {
 			flowProps.put(prop.getSource(), prop);
 		}
 	}
-
+	
 	public String getId() {
 		return id;
 	}

diff --git a/src/java/azkaban/flow/Node.java b/src/java/azkaban/flow/Node.java
index 62d4737..fbe5a39 100644
--- a/src/java/azkaban/flow/Node.java
+++ b/src/java/azkaban/flow/Node.java
@@ -23,7 +23,6 @@ import java.util.Map;
 import azkaban.utils.Utils;
 
 public class Node {
-
 	private final String id;
 	private String jobSource;
 	private String propsSource;

diff --git a/src/java/azkaban/flow/SpecialJobTypes.java b/src/java/azkaban/flow/SpecialJobTypes.java
new file mode 100644
index 0000000..dfbe03a
--- /dev/null
+++ b/src/java/azkaban/flow/SpecialJobTypes.java
@@ -0,0 +1,9 @@
+package azkaban.flow;
+
+public class SpecialJobTypes {
+	public static final String BRANCH_START_TYPE = "branch.start";
+	public static final String BRANCH_END_TYPE = "branch.end";
+
+	public static final String EMBEDDED_FLOW_TYPE = "flow";
+	public static final String FLOW_NAME = "flow.name";
+}

diff --git a/src/java/azkaban/utils/DirectoryFlowLoader.java b/src/java/azkaban/utils/DirectoryFlowLoader.java
index d0f14fc..bad673c 100644
--- a/src/java/azkaban/utils/DirectoryFlowLoader.java
+++ b/src/java/azkaban/utils/DirectoryFlowLoader.java
@@ -19,6 +19,7 @@ import azkaban.flow.Edge;
 import azkaban.flow.Flow;
 import azkaban.flow.FlowProps;
 import azkaban.flow.Node;
+import azkaban.flow.SpecialJobTypes;
 
 public class DirectoryFlowLoader {
 	private static final DirFilter DIR_FILTER = new DirFilter();
@@ -26,15 +27,20 @@ public class DirectoryFlowLoader {
 	private static final String JOB_SUFFIX = ".job";
 	
 	private final Logger logger;
+	private HashSet<String> rootNodes;
 	private HashMap<String, Flow> flowMap;
 	private HashMap<String, Node> nodeMap;
 	private HashMap<String, Map<String, Edge>> nodeDependencies;
 	private HashMap<String, Props> jobPropsMap;
+	
+	// Flow dependencies for embedded flows.
+	private HashMap<String, Set<String>> flowDependencies;
+	
 	private ArrayList<FlowProps> flowPropsList;
 	private ArrayList<Props> propsList;
 	private Set<String> errors;
 	private Set<String> duplicateJobs;
-	
+
 	public DirectoryFlowLoader(Logger logger) {
 		this.logger = logger;
 	}
@@ -64,16 +70,19 @@ public class DirectoryFlowLoader {
 		errors = new HashSet<String>();
 		duplicateJobs = new HashSet<String>();
 		nodeDependencies = new HashMap<String, Map<String, Edge>>();
+		rootNodes = new HashSet<String>();
 
 		// Load all the props files and create the Node objects
 		loadProjectFromDir(baseDirectory.getPath(), baseDirectory, null);
 		
 		// Create edges and find missing dependencies
 		resolveDependencies();
-
+		
 		// Create the flows.
 		buildFlowsFromDependencies();
-
+		
+		// Resolve embedded flows
+		resolveEmbeddedFlows();
 	}
 	
 	private void loadProjectFromDir(String base, File dir, Props parent) {
@@ -96,7 +105,6 @@ public class DirectoryFlowLoader {
 			propsList.add(parent);
 		}
 		
-		
 		// Load all Job files. If there's a duplicate name, then we don't load
 		File[] jobFiles = dir.listFiles(new SuffixFilter(JOB_SUFFIX));
 		for (File file: jobFiles) {
@@ -127,11 +135,15 @@ public class DirectoryFlowLoader {
 							node.setPropsSource(parent.getSource());
 						}
 
+						// Force root node
+						if(prop.getBoolean(CommonJobProperties.ROOT_NODE, false)) {
+							rootNodes.add(jobName);
+						}
+						
 						jobPropsMap.put(jobName, prop);
 						nodeMap.put(jobName, node);
 					}
 				}
-				
 			} catch (IOException e) {
 				errors.add("Error loading job file " + file.getName() + ":" + e.getMessage());
 			}
@@ -143,6 +155,34 @@ public class DirectoryFlowLoader {
 		}
 	}
 	
+	private void resolveEmbeddedFlows() {
+		for (String flowId: flowDependencies.keySet()) {
+			HashSet<String> visited = new HashSet<String>();
+			resolveEmbeddedFlow(flowId, visited);
+		}
+	}
+	
+	private void resolveEmbeddedFlow(String flowId, Set<String> visited) {
+		visited.add(flowId);
+		
+		Set<String> embeddedFlow = flowDependencies.get(flowId);
+		for (String embeddedFlowId: embeddedFlow) {
+			if (visited.contains(embeddedFlowId)) {
+				errors.add("Embedded flow cycle found in " + flowId + "->" + embeddedFlowId);
+				return;
+			}
+			else if (!flowMap.containsKey(embeddedFlowId)) {
+				errors.add("Flow " + flowId + " depends on " + embeddedFlowId + " but can't be found.");
+				return;
+			}
+			else {
+				resolveEmbeddedFlow(embeddedFlowId, visited);
+			}
+		}
+		
+		visited.remove(flowId);
+	}
+	
 	private void resolveDependencies() {
 		// Add all the in edges and out edges. Catch bad dependencies and self referrals. Also collect list of nodes who are parents.
 		for (Node node: nodeMap.values()) {
@@ -211,7 +251,9 @@ public class DirectoryFlowLoader {
 		// Now create flows. Bad flows are marked invalid
 		Set<String> visitedNodes = new HashSet<String>();
 		for (Node base: nodeMap.values()) {
-			if (!nonRootNodes.contains(base.getId())) {
+			// Root nodes can be discovered when parsing jobs
+			if (rootNodes.contains(base.getId()) || !nonRootNodes.contains(base.getId())) {
+				rootNodes.add(base.getId());
 				Flow flow = new Flow(base.getId());
 				Props jobProp = jobPropsMap.get(base.getId());
 				
@@ -252,8 +294,19 @@ public class DirectoryFlowLoader {
 	private void constructFlow(Flow flow, Node node, Set<String> visited) {
 		visited.add(node.getId());
 
-		// Clone the node so each flow can operate on its own node
 		flow.addNode(node);
+		if (SpecialJobTypes.EMBEDDED_FLOW_TYPE.equals(node.getType())) {
+			Props props = jobPropsMap.get(node.getId());
+			String embeddedFlow = props.get(SpecialJobTypes.FLOW_NAME);
+			
+			Set<String> embeddedFlows = flowDependencies.get(flow.getId());
+			if (embeddedFlows == null) {
+				embeddedFlows = new HashSet<String>();
+				flowDependencies.put(flow.getId(), embeddedFlows);
+			}
+
+			embeddedFlows.add(embeddedFlow);
+		}
 		Map<String, Edge> dependencies = nodeDependencies.get(node.getId());
 
 		if (dependencies != null) {
@@ -279,7 +332,7 @@ public class DirectoryFlowLoader {
 
 		visited.remove(node.getId());
 	}
-
+	
 	private String getNameWithoutExtension(File file) {
 		String filename = file.getName();
 		int index = filename.lastIndexOf('.');

diff --git a/src/java/azkaban/utils/PropsUtils.java b/src/java/azkaban/utils/PropsUtils.java
index 16afb52..5324cb7 100644
--- a/src/java/azkaban/utils/PropsUtils.java
+++ b/src/java/azkaban/utils/PropsUtils.java
@@ -24,7 +24,6 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.StringTokenizer;
 import java.util.UUID;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -145,8 +144,7 @@ public class PropsUtils {
 		return false;
 	}
 
-	private static final Pattern VARIABLE_PATTERN = Pattern
-			.compile("\\$\\{([a-zA-Z_.0-9]+)\\}");
+	private static final Pattern VARIABLE_PATTERN = Pattern.compile("\\$\\{([a-zA-Z_.0-9]+)\\}");
 
 	public static Props resolveProps(Props props) {
 		if (props == null) return null;

diff --git a/unit/build.xml b/unit/build.xml
index b35b866..cb07807 100644
--- a/unit/build.xml
+++ b/unit/build.xml
@@ -9,7 +9,8 @@
 	<property name="java.src.dir" value="${base.dir}/unit/java" />
 	<property name="job.conf.dir" value="${base.dir}/unit/executions/exectest1" />
 	<property name="job.conf.dir2" value="${base.dir}/unit/executions/exectest2" />
-	
+	<property name="animal.conf.dir" value="${base.dir}/unit/executions/animal" />
+		
 	<property environment="env" />
 
 	<path id="main.classpath">
@@ -74,6 +75,16 @@
 			<zipfileset dir="${job.conf.dir2}" />
 		</zip>
 	</target>
-
+	
+	<target name="package-animal" depends="jars" description="Creates a test zip">
+		<delete dir="${dist.packages.dir}" />
+		<mkdir dir="${dist.packages.dir}" />
+		
+		<!-- Tarball it -->
+		<zip destfile="${dist.packages.dir}/animals.zip">
+			<zipfileset dir="${dist.jar.dir}" />
+			<zipfileset dir="${animal.conf.dir}" />
+		</zip>
+	</target>
 	
 </project>

diff --git a/unit/executions/animal/albatross.job b/unit/executions/animal/albatross.job
new file mode 100644
index 0000000..4f9cf95
--- /dev/null
+++ b/unit/executions/animal/albatross.job
@@ -0,0 +1,4 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=21
+fail=false
\ No newline at end of file

diff --git a/unit/executions/animal/animals.job b/unit/executions/animal/animals.job
new file mode 100644
index 0000000..8d2e4a4
--- /dev/null
+++ b/unit/executions/animal/animals.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=11
+fail=false
+dependencies=humpback-whale
\ No newline at end of file

diff --git a/unit/executions/animal/baboon.job b/unit/executions/animal/baboon.job
new file mode 100644
index 0000000..0a4e652
--- /dev/null
+++ b/unit/executions/animal/baboon.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=7
+fail=false
+dependencies=albatross
\ No newline at end of file

diff --git a/unit/executions/animal/caiman.job b/unit/executions/animal/caiman.job
new file mode 100644
index 0000000..2e258ac
--- /dev/null
+++ b/unit/executions/animal/caiman.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=13
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/camel.job b/unit/executions/animal/camel.job
new file mode 100644
index 0000000..a854a97
--- /dev/null
+++ b/unit/executions/animal/camel.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=18
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/capybara.job b/unit/executions/animal/capybara.job
new file mode 100644
index 0000000..02db6f5
--- /dev/null
+++ b/unit/executions/animal/capybara.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=16
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/cat.job b/unit/executions/animal/cat.job
new file mode 100644
index 0000000..f7de0a7
--- /dev/null
+++ b/unit/executions/animal/cat.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=23
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/caterpillar.job b/unit/executions/animal/caterpillar.job
new file mode 100644
index 0000000..6b4347c
--- /dev/null
+++ b/unit/executions/animal/caterpillar.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=19
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/catfish.job b/unit/executions/animal/catfish.job
new file mode 100644
index 0000000..538d93a
--- /dev/null
+++ b/unit/executions/animal/catfish.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=14
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/centipede.job b/unit/executions/animal/centipede.job
new file mode 100644
index 0000000..2e258ac
--- /dev/null
+++ b/unit/executions/animal/centipede.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=13
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/chameleon.job b/unit/executions/animal/chameleon.job
new file mode 100644
index 0000000..7393129
--- /dev/null
+++ b/unit/executions/animal/chameleon.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=25
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/cheetah.job b/unit/executions/animal/cheetah.job
new file mode 100644
index 0000000..5003eb3
--- /dev/null
+++ b/unit/executions/animal/cheetah.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=7
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/chicken.job b/unit/executions/animal/chicken.job
new file mode 100644
index 0000000..dbbbc72
--- /dev/null
+++ b/unit/executions/animal/chicken.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=29
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/chihuahua.job b/unit/executions/animal/chihuahua.job
new file mode 100644
index 0000000..a854a97
--- /dev/null
+++ b/unit/executions/animal/chihuahua.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=18
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/clown-fish.job b/unit/executions/animal/clown-fish.job
new file mode 100644
index 0000000..538d93a
--- /dev/null
+++ b/unit/executions/animal/clown-fish.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=14
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/cockroach.job b/unit/executions/animal/cockroach.job
new file mode 100644
index 0000000..7449537
--- /dev/null
+++ b/unit/executions/animal/cockroach.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=9
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/cougar.job b/unit/executions/animal/cougar.job
new file mode 100644
index 0000000..d29ca57
--- /dev/null
+++ b/unit/executions/animal/cougar.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=30
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/cuttlefish.job b/unit/executions/animal/cuttlefish.job
new file mode 100644
index 0000000..dbbbc72
--- /dev/null
+++ b/unit/executions/animal/cuttlefish.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=29
+fail=false
+dependencies=baboon
\ No newline at end of file

diff --git a/unit/executions/animal/dolphin.job b/unit/executions/animal/dolphin.job
new file mode 100644
index 0000000..a1b6a0c
--- /dev/null
+++ b/unit/executions/animal/dolphin.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=14
+fail=false
+dependencies=caterpillar,chameleon,cougar,camel,cuttlefish,centipede,cheetah
\ No newline at end of file

diff --git a/unit/executions/animal/elephant.job b/unit/executions/animal/elephant.job
new file mode 100644
index 0000000..d2b6853
--- /dev/null
+++ b/unit/executions/animal/elephant.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=8
+fail=false
+dependencies=camel,caiman,capybara,cat,catfish,chicken,chihuahua,clown-fish,cockroach,dolphin
\ No newline at end of file

diff --git a/unit/executions/animal/flamingo.job b/unit/executions/animal/flamingo.job
new file mode 100644
index 0000000..07b68a9
--- /dev/null
+++ b/unit/executions/animal/flamingo.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=7
+fail=false
+dependencies=camel,elephant
\ No newline at end of file

diff --git a/unit/executions/animal/gorilla.job b/unit/executions/animal/gorilla.job
new file mode 100644
index 0000000..f5b16e7
--- /dev/null
+++ b/unit/executions/animal/gorilla.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=27
+fail=false
+dependencies=flamingo
\ No newline at end of file

diff --git a/unit/executions/animal/humpback-whale.job b/unit/executions/animal/humpback-whale.job
new file mode 100644
index 0000000..4981525
--- /dev/null
+++ b/unit/executions/animal/humpback-whale.job
@@ -0,0 +1,5 @@
+type=java
+job.class=azkaban.test.executor.SleepJavaJob
+seconds=19
+fail=false
+dependencies=gorilla
\ No newline at end of file

diff --git a/unit/java/azkaban/Scrubber.java b/unit/java/azkaban/Scrubber.java
new file mode 100644
index 0000000..17c7d65
--- /dev/null
+++ b/unit/java/azkaban/Scrubber.java
@@ -0,0 +1,20 @@
+package azkaban;
+
+import java.io.File;
+
+import org.apache.log4j.Logger;
+
+import azkaban.utils.DirectoryFlowLoader;
+
+public class Scrubber {
+	private static Logger logger = Logger.getLogger(Scrubber.class);
+	
+	public static void main(String[] args) {
+		DirectoryFlowLoader loader = new DirectoryFlowLoader(logger);
+		
+		File baseDir = new File(args[0]);
+		loader.loadProjectFlow(baseDir);
+	
+		loader.getFlowMap();
+	}
+}
\ No newline at end of file