@Test
public void testForkMonitoringSeries() {

  /*
   * Setup the pipeline of operation processors
   */
  List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>();
  List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>();
  fork1.add(new OperationProcessor(new DummyThrottleOperationFactory()));
  fork1.add(new OperationProcessor(new DummyOperationFactory()));
  forks.add(fork1);
  ForkOperation op = new ForkOperation(forks);
  /*
   * Create thread that supplies input events
   */
  Queue<InternalEvent> inputQueue = new Queue<InternalEvent>();
  supply(10, inputQueue);
  /*
   * Process
   */
  Stream<InternalEvent> input = inputQueue.stream();
  Stream<InternalEvent> output = op.getOutputStream(input);
  List<String> outputEvents =
      output.map(InternalEvent::getEventString).collect(Collectors.toList());
  Monitor m = Monitor.getInstance();
  List<Stat> stats = m.getStats();
  Map<String, String> actual = new HashMap<String, String>();
  for (Stat stat : stats) {
    String key =
        String.format("%s.%s", stat.getTags().iterator().next().getValue(), stat.getName());
    actual.put(key, "" + stat.getValue());
  }
  assertEquals("0", actual
      .get("com.nextdoor.bender.testutils.DummyOperationHelper.DummyOperation.error.count"));
  assertEquals("10", actual
      .get("com.nextdoor.bender.testutils.DummyOperationHelper.DummyOperation.success.count"));
  assertEquals("10", actual.get(
      "com.nextdoor.bender.testutils.DummyThrottleOperationHelper.DummyThrottleOperation.success.count"));
  assertEquals("0", actual.get(
      "com.nextdoor.bender.testutils.DummyThrottleOperationHelper.DummyThrottleOperation.error.count"));
}