@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 testPerElementProcessingTimeCounters() throws Exception {
    PipelineOptions options = PipelineOptionsFactory.create();
    options
        .as(DataflowPipelineDebugOptions.class)
        .setExperiments(
            Lists.newArrayList(DataflowElementExecutionTracker.TIME_PER_ELEMENT_EXPERIMENT));
    ExecutionStateSampler stateSampler = ExecutionStateSampler.newForTest();
    DataflowExecutionStateTracker stateTracker =
        new DataflowExecutionStateTracker(
            stateSampler,
            new TestDataflowExecutionState(
                NameContext.forStage("test-stage"),
                "other",
                null /* requestingStepName */,
                null /* sideInputIndex */,
                null /* metricsContainer */,
                NoopProfileScope.NOOP),
            counterSet,
            options,
            "test-work-item-id");
    NameContext parDoName = nameForStep("s1");

    // Wire a read operation with 3 elements to a ParDoOperation and assert that we count
    // the correct number of elements.
    ReadOperation read =
        ReadOperation.forTest(
            new TestReader("a", "b", "c"),
            new OutputReceiver(),
            TestOperationContext.create(counterSet, nameForStep("s0"), null, stateTracker));
    ParDoOperation parDo =
        new ParDoOperation(
            new NoopParDoFn(),
            new OutputReceiver[0],
            TestOperationContext.create(counterSet, parDoName, null, stateTracker));
    parDo.attachInput(read, 0);
    List<Operation> operations = Lists.newArrayList(read, parDo);

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

    CounterName counterName =
        CounterName.named("per-element-processing-time").withOriginalName(parDoName);
    Counter<Long, CounterDistribution> counter =
        (Counter<Long, CounterDistribution>) counterSet.getExistingCounter(counterName);

    assertThat(counter.getAggregate().getCount(), equalTo(3L));
  }