@Override
  public void execute() throws Exception {
    LOG.debug("Executing map task");

    try (Closeable stateCloser = executionStateTracker.activate()) {
      try {
        // Start operations, in reverse-execution-order, so that a
        // consumer is started before a producer might output to it.
        // Starting a root operation such as a ReadOperation does the work
        // of processing the input dataset.
        LOG.debug("Starting operations");
        ListIterator<Operation> iterator = operations.listIterator(operations.size());
        while (iterator.hasPrevious()) {
          if (Thread.currentThread().isInterrupted()) {
            throw new InterruptedException("Worker aborted");
          }
          Operation op = iterator.previous();
          op.start();
        }

        // Finish operations, in forward-execution-order, so that a
        // producer finishes outputting to its consumers before those
        // consumers are themselves finished.
        LOG.debug("Finishing operations");
        for (Operation op : operations) {
          if (Thread.currentThread().isInterrupted()) {
            throw new InterruptedException("Worker aborted");
          }
          op.finish();
        }
      } catch (Exception | Error exn) {
        LOG.debug("Aborting operations", exn);
        for (Operation op : operations) {
          try {
            op.abort();
          } catch (Exception | Error exn2) {
            exn.addSuppressed(exn2);
            if (exn2 instanceof InterruptedException) {
              Thread.currentThread().interrupt();
            }
          }
        }
        throw exn;
      }
    }

    LOG.debug("Map task execution complete");

    // TODO: support for success / failure ports?
  }
@Test
  public void testExecuteMapTaskExecutor() throws Exception {

    Operation o1 = Mockito.mock(Operation.class);
    Operation o2 = Mockito.mock(Operation.class);
    Operation o3 = Mockito.mock(Operation.class);
    List<Operation> operations = Arrays.asList(new Operation[] {o1, o2, o3});

    ExecutionStateTracker stateTracker = Mockito.spy(ExecutionStateTracker.newForTest());

    try (MapTaskExecutor executor = new MapTaskExecutor(operations, counterSet, stateTracker)) {
      executor.execute();
    }

    InOrder inOrder = Mockito.inOrder(stateTracker, o1, o2, o3);
    inOrder.verify(stateTracker).activate();
    inOrder.verify(o3).start();
    inOrder.verify(o2).start();
    inOrder.verify(o1).start();
    inOrder.verify(o1).finish();
    inOrder.verify(o2).finish();
    inOrder.verify(o3).finish();
    inOrder.verify(stateTracker).deactivate();
  }