azkaban-aplcache

diff --git a/azkaban-exec-server/src/test/java/azkaban/execapp/EventCollectorListener.java b/azkaban-exec-server/src/test/java/azkaban/execapp/EventCollectorListener.java
index e4253c5..72eefa9 100644
--- a/azkaban-exec-server/src/test/java/azkaban/execapp/EventCollectorListener.java
+++ b/azkaban-exec-server/src/test/java/azkaban/execapp/EventCollectorListener.java
@@ -16,6 +16,9 @@
 
 package azkaban.execapp;
 
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
 import azkaban.event.Event;
 import azkaban.event.Event.Type;
 import azkaban.event.EventListener;
@@ -49,13 +52,6 @@ public class EventCollectorListener implements EventListener {
     return this.eventList;
   }
 
-  public void writeAllEvents() {
-    for (final Event event : this.eventList) {
-      System.out.print(event.getType());
-      System.out.print(",");
-    }
-  }
-
   public boolean checkOrdering() {
     final long time = 0;
     for (final Event event : this.eventList) {
@@ -67,27 +63,11 @@ public class EventCollectorListener implements EventListener {
     return true;
   }
 
-  public void checkEventExists(final Type[] types) {
-    int index = 0;
-    for (final Event event : this.eventList) {
-      if (event.getRunner() == null) {
-        continue;
-      }
-
-      if (index >= types.length) {
-        throw new RuntimeException("More events than expected. Got "
-            + event.getType());
-      }
-      final Type type = types[index++];
-
-      if (type != event.getType()) {
-        throw new RuntimeException("Got " + event.getType() + ", expected "
-            + type + " index:" + index);
-      }
-    }
-
-    if (types.length != index) {
-      throw new RuntimeException("Not enough events.");
-    }
+  public void checkEventExists(final Type... expected) {
+    Object[] captured = this.eventList.stream()
+        .filter(event -> event.getRunner() != null)
+        .map(event -> event.getType())
+        .toArray();
+    assertThat(captured, is(expected));
   }
 }

diff --git a/azkaban-exec-server/src/test/java/azkaban/execapp/FlowRunnerTest.java b/azkaban-exec-server/src/test/java/azkaban/execapp/FlowRunnerTest.java
index 11709c6..f38e8b1 100644
--- a/azkaban-exec-server/src/test/java/azkaban/execapp/FlowRunnerTest.java
+++ b/azkaban-exec-server/src/test/java/azkaban/execapp/FlowRunnerTest.java
@@ -117,14 +117,7 @@ public class FlowRunnerTest extends FlowRunnerTestBase {
     assertStatus("job8", Status.SUCCEEDED);
     assertStatus("job10", Status.SUCCEEDED);
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.FLOW_STARTED,
-          Type.FLOW_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.FLOW_STARTED, Type.FLOW_FINISHED);
   }
 
   @Test
@@ -163,14 +156,7 @@ public class FlowRunnerTest extends FlowRunnerTestBase {
     assertStatus("job8", Status.SUCCEEDED);
     assertStatus("job10", Status.SKIPPED);
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.FLOW_STARTED,
-          Type.FLOW_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.FLOW_STARTED, Type.FLOW_FINISHED);
   }
 
   @Test
@@ -200,14 +186,7 @@ public class FlowRunnerTest extends FlowRunnerTestBase {
     assertStatus("job10", Status.CANCELLED);
     assertThreadShutDown();
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.FLOW_STARTED,
-          Type.FLOW_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.FLOW_STARTED, Type.FLOW_FINISHED);
   }
 
   @Test
@@ -238,14 +217,7 @@ public class FlowRunnerTest extends FlowRunnerTestBase {
     assertStatus("job9", Status.CANCELLED);
     assertStatus("job10", Status.CANCELLED);
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.FLOW_STARTED,
-          Type.FLOW_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-      eventCollector.writeAllEvents();
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.FLOW_STARTED, Type.FLOW_FINISHED);
   }
 
   @Test
@@ -275,14 +247,7 @@ public class FlowRunnerTest extends FlowRunnerTestBase {
     assertStatus("job10", Status.CANCELLED);
     assertThreadShutDown();
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.FLOW_STARTED,
-          Type.FLOW_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-      eventCollector.writeAllEvents();
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.FLOW_STARTED, Type.FLOW_FINISHED);
   }
 
   @Test
@@ -312,14 +277,7 @@ public class FlowRunnerTest extends FlowRunnerTestBase {
 
     assertFlowStatus(Status.KILLED);
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.FLOW_STARTED,
-          Type.FLOW_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-      eventCollector.writeAllEvents();
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.FLOW_STARTED, Type.FLOW_FINISHED);
   }
 
   @Test

diff --git a/azkaban-exec-server/src/test/java/azkaban/execapp/JobRunnerTest.java b/azkaban-exec-server/src/test/java/azkaban/execapp/JobRunnerTest.java
index 49e02eb..8aec1a4 100644
--- a/azkaban-exec-server/src/test/java/azkaban/execapp/JobRunnerTest.java
+++ b/azkaban-exec-server/src/test/java/azkaban/execapp/JobRunnerTest.java
@@ -102,13 +102,7 @@ public class JobRunnerTest {
 
     Assert.assertTrue(loader.getNodeUpdateCount(node.getId()) == 3);
 
-    Assert.assertTrue(eventCollector.checkOrdering());
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_STARTED,
-          Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_STARTED, Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED);
   }
 
   @Ignore
@@ -137,12 +131,7 @@ public class JobRunnerTest {
     Assert.assertTrue(!runner.isKilled());
     Assert.assertTrue(loader.getNodeUpdateCount(node.getId()) == 3);
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_STARTED,
-          Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_STARTED, Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED);
   }
 
   @Test
@@ -173,12 +162,7 @@ public class JobRunnerTest {
 
     Assert.assertTrue(loader.getNodeUpdateCount(node.getId()) == null);
 
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_STARTED,
-          Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_STARTED, Type.JOB_FINISHED);
   }
 
   @Test
@@ -209,12 +193,7 @@ public class JobRunnerTest {
     Assert.assertTrue(outputProps == null);
     Assert.assertTrue(runner.getLogFilePath() == null);
     Assert.assertTrue(!runner.isKilled());
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_STARTED,
-          Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_STARTED, Type.JOB_FINISHED);
   }
 
   @Ignore
@@ -253,14 +232,7 @@ public class JobRunnerTest {
     Assert.assertTrue(logFile.exists());
     Assert.assertTrue(eventCollector.checkOrdering());
     Assert.assertTrue(runner.isKilled());
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_STARTED,
-          Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      System.out.println(e.getMessage());
-
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_STARTED, Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED);
   }
 
   @Ignore
@@ -295,12 +267,7 @@ public class JobRunnerTest {
     Assert.assertTrue(loader.getNodeUpdateCount(node.getId()) == 3);
 
     Assert.assertTrue(eventCollector.checkOrdering());
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_STARTED,
-          Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_STARTED, Type.JOB_STATUS_CHANGED, Type.JOB_FINISHED);
   }
 
   @Test
@@ -339,12 +306,7 @@ public class JobRunnerTest {
     Assert.assertTrue(outputProps == null);
     Assert.assertTrue(logFile.exists());
 
-    Assert.assertTrue(eventCollector.checkOrdering());
-    try {
-      eventCollector.checkEventExists(new Type[]{Type.JOB_FINISHED});
-    } catch (final Exception e) {
-      Assert.fail(e.getMessage());
-    }
+    eventCollector.checkEventExists(Type.JOB_FINISHED);
   }
 
   private Props createProps(final int sleepSec, final boolean fail) {