@Override
  public void abort() {
    // Signal the read loop to abort on the next record and async abort any iterators.
    // TODO: Also interrupt the execution thread.
    for (Operation op : operations) {
      Preconditions.checkState(op instanceof ReadOperation || op instanceof ReceivingOperation);
      if (op instanceof ReadOperation) {
        ((ReadOperation) op).abortReadLoop();
      }
    }
  }
@Test
  public void testAbort() throws Exception {
    // Operation must be an instance of ReadOperation or ReceivingOperation per preconditions
    // in MapTaskExecutor.
    ReadOperation o1 = Mockito.mock(ReadOperation.class);
    ReadOperation o2 = Mockito.mock(ReadOperation.class);

    ExecutionStateTracker stateTracker = Mockito.spy(ExecutionStateTracker.newForTest());
    MapTaskExecutor executor =
        new MapTaskExecutor(Arrays.<Operation>asList(o1, o2), counterSet, stateTracker);
    Mockito.doAnswer(
            invocation -> {
              executor.abort();
              return null;
            })
        .when(o1)
        .finish();
    executor.execute();
    Mockito.verify(stateTracker).activate();
    Mockito.verify(o1, atLeastOnce()).abortReadLoop();
    Mockito.verify(o2, atLeastOnce()).abortReadLoop();
    Mockito.verify(stateTracker).deactivate();
  }