public ReadOperation getReadOperation() {
    if (operations == null || operations.isEmpty()) {
      throw new IllegalStateException("Map task has no operation.");
    }

    Operation readOperation = operations.get(0);
    if (!(readOperation instanceof ReadOperation)) {
      throw new IllegalStateException("First operation in the map task is not a ReadOperation.");
    }

    return (ReadOperation) readOperation;
  }
@Test
  public void testNoOperation() throws Exception {
    // Test MapTaskExecutor without a single operation.
    ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest();
    try (MapTaskExecutor executor =
        new MapTaskExecutor(new ArrayList<Operation>(), counterSet, stateTracker)) {
      thrown.expect(IllegalStateException.class);
      thrown.expectMessage("has no operation");
      executor.getReadOperation();
    }
  }