public static IntrinsicMapTaskExecutor withSharedCounterSet(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker) {
    return new IntrinsicMapTaskExecutor(operations, counters, executionStateTracker);
  }
@Test
  @SuppressWarnings("unchecked")
  public void testGetOutputCounters() throws Exception {
    List<Operation> operations =
        Arrays.asList(
            new Operation[] {
              createOperation("o1", 1), createOperation("o2", 2), createOperation("o3", 3)
            });

    ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest();
    try (IntrinsicMapTaskExecutor executor =
        IntrinsicMapTaskExecutor.withSharedCounterSet(operations, counterSet, stateTracker)) {
      CounterSet counterSet = executor.getOutputCounters();

      CounterUpdateExtractor<?> updateExtractor = Mockito.mock(CounterUpdateExtractor.class);
      counterSet.extractUpdates(false, updateExtractor);
      verify(updateExtractor).longSum(eq(named("test-o1-ElementCount")), anyBoolean(), eq(1L));
      verify(updateExtractor).longSum(eq(named("test-o2-ElementCount")), anyBoolean(), eq(2L));
      verify(updateExtractor).longSum(eq(named("test-o3-ElementCount")), anyBoolean(), eq(3L));
      verifyNoMoreInteractions(updateExtractor);
    }
  }