MapTaskExecutor implements WorkExecutor { @Override public CounterSet getOutputCounters() { return counters; } MapTaskExecutor(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker); @Override CounterSet getOutputCounters(); @Override void execute(); @Override NativeReader.Progress getWorkerProgress(); @Override NativeReader.@Nullable DynamicSplitResult requestCheckpoint(); @Override NativeReader.DynamicSplitResult requestDynamicSplit(
      NativeReader.DynamicSplitRequest splitRequest); ReadOperation getReadOperation(); boolean supportsRestart(); @Override void abort(); @Override List<Integer> reportProducedEmptyOutput(); final List<Operation> operations; }
@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 (MapTaskExecutor executor = new MapTaskExecutor(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); } }