MapTaskExecutor implements WorkExecutor { public ReadOperation getReadOperation() throws Exception { 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; } 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 public void testNoOperation() throws Exception { 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(); } }
@Test public void testNoReadOperation() throws Exception { List<Operation> operations = Arrays.<Operation>asList(createOperation("o1", 1), createOperation("o2", 2)); ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (MapTaskExecutor executor = new MapTaskExecutor(operations, counterSet, stateTracker)) { thrown.expect(IllegalStateException.class); thrown.expectMessage("is not a ReadOperation"); executor.getReadOperation(); } }
@Test public void testValidOperations() throws Exception { TestOutputReceiver receiver = new TestOutputReceiver(counterSet, NameContextsForTests.nameContextForTest()); List<Operation> operations = Arrays.<Operation>asList(new TestReadOperation(receiver, createContext("ReadOperation"))); ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (MapTaskExecutor executor = new MapTaskExecutor(operations, counterSet, stateTracker)) { Assert.assertEquals(operations.get(0), executor.getReadOperation()); } }