IntrinsicMapTaskExecutor extends DataflowMapTaskExecutor { public static IntrinsicMapTaskExecutor withSharedCounterSet( List<Operation> operations, CounterSet counters, ExecutionStateTracker executionStateTracker) { return new IntrinsicMapTaskExecutor(operations, counters, executionStateTracker); } @Deprecated protected  IntrinsicMapTaskExecutor(
      List<Operation> operations,
      CounterSet counterSet,
      ExecutionStateTracker executionStateTracker); static IntrinsicMapTaskExecutor forOperations(
      List<Operation> operations, ExecutionStateTracker executionStateTracker); static IntrinsicMapTaskExecutor withSharedCounterSet(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker); @Override Iterable<CounterUpdate> extractMetricUpdates();  }
@Test public void testExecuteMapTaskExecutor() throws Exception { List<String> log = new ArrayList<>(); Operation o1 = Mockito.mock(Operation.class); Operation o2 = Mockito.mock(Operation.class); Operation o3 = Mockito.mock(Operation.class); List<Operation> operations = Arrays.asList(new Operation[] {o1, o2, o3}); ExecutionStateTracker stateTracker = Mockito.mock(ExecutionStateTracker.class); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet(operations, counterSet, stateTracker)) { executor.execute(); } InOrder inOrder = Mockito.inOrder(stateTracker, o1, o2, o3); inOrder.verify(o3).start(); inOrder.verify(o2).start(); inOrder.verify(o1).start(); inOrder.verify(o1).finish(); inOrder.verify(o2).finish(); inOrder.verify(o3).finish(); }
@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); } }
@Test public void testPerElementProcessingTimeCounters() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); options .as(DataflowPipelineDebugOptions.class) .setExperiments( Lists.newArrayList(DataflowElementExecutionTracker.TIME_PER_ELEMENT_EXPERIMENT)); DataflowExecutionStateTracker stateTracker = new DataflowExecutionStateTracker( ExecutionStateSampler.newForTest(), new TestDataflowExecutionState( NameContext.forStage("test-stage"), "other", null , null , null , NoopProfileScope.NOOP), counterSet, options, "test-work-item-id"); NameContext parDoName = nameForStep("s1"); ReadOperation read = ReadOperation.forTest( new TestReader("a", "b", "c"), new OutputReceiver(), TestOperationContext.create(counterSet, nameForStep("s0"), null, stateTracker)); ParDoOperation parDo = new ParDoOperation( new NoopParDoFn(), new OutputReceiver[0], TestOperationContext.create(counterSet, parDoName, null, stateTracker)); parDo.attachInput(read, 0); List<Operation> operations = Lists.newArrayList(read, parDo); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet(operations, counterSet, stateTracker)) { executor.execute(); } CounterName counterName = CounterName.named("per-element-processing-time").withOriginalName(parDoName); Counter<Long, CounterDistribution> counter = (Counter<Long, CounterDistribution>) counterSet.getExistingCounter(counterName); assertThat(counter.getAggregate().getCount(), equalTo(3L)); }
@Test @SuppressWarnings("unchecked") public void testGetMetricContainers() throws Exception { ExecutionStateTracker stateTracker = new DataflowExecutionStateTracker( ExecutionStateSampler.newForTest(), new TestDataflowExecutionState( NameContext.forStage("testStage"), "other", null , null , null , NoopProfileScope.NOOP), new CounterSet(), PipelineOptionsFactory.create(), "test-work-item-id"); final String o1 = "o1"; TestOperationContext context1 = createContext(o1, stateTracker); final String o2 = "o2"; TestOperationContext context2 = createContext(o2, stateTracker); final String o3 = "o3"; TestOperationContext context3 = createContext(o3, stateTracker); List<Operation> operations = Arrays.asList( new Operation(new OutputReceiver[] {}, context1) { @Override public void start() throws Exception { super.start(); try (Closeable scope = context.enterStart()) { Metrics.counter("TestMetric", "MetricCounter").inc(1L); } } }, new Operation(new OutputReceiver[] {}, context2) { @Override public void start() throws Exception { super.start(); try (Closeable scope = context.enterStart()) { Metrics.counter("TestMetric", "MetricCounter").inc(2L); } } }, new Operation(new OutputReceiver[] {}, context3) { @Override public void start() throws Exception { super.start(); try (Closeable scope = context.enterStart()) { Metrics.counter("TestMetric", "MetricCounter").inc(3L); } } }); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet(operations, counterSet, stateTracker)) { executor.execute(); assertThat( context1.metricsContainer().getUpdates().counterUpdates(), contains(metricUpdate("TestMetric", "MetricCounter", o1, 1L))); assertThat( context2.metricsContainer().getUpdates().counterUpdates(), contains(metricUpdate("TestMetric", "MetricCounter", o2, 2L))); assertThat( context3.metricsContainer().getUpdates().counterUpdates(), contains(metricUpdate("TestMetric", "MetricCounter", o3, 3L))); } }
@Test public void testNoOperation() throws Exception { ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet( 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 (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet(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 (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet(operations, counterSet, stateTracker)) { Assert.assertEquals(operations.get(0), executor.getReadOperation()); } }
@Test public void testGetProgressAndRequestSplit() throws Exception { TestOutputReceiver receiver = new TestOutputReceiver(counterSet, NameContextsForTests.nameContextForTest()); TestReadOperation operation = new TestReadOperation(receiver, createContext("ReadOperation")); ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet( Arrays.asList(new Operation[] {operation}), counterSet, stateTracker)) { operation.setProgress(approximateProgressAtIndex(1L)); Assert.assertEquals(positionAtIndex(1L), positionFromProgress(executor.getWorkerProgress())); Assert.assertEquals( positionAtIndex(1L), positionFromSplitResult(executor.requestDynamicSplit(splitRequestAtIndex(1L)))); } }
@Test public void testExceptionInStartAbortsAllOperations() throws Exception { Operation o1 = Mockito.mock(Operation.class); Operation o2 = Mockito.mock(Operation.class); Operation o3 = Mockito.mock(Operation.class); Mockito.doThrow(new Exception("in start")).when(o2).start(); ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet( Arrays.<Operation>asList(o1, o2, o3), counterSet, stateTracker)) { executor.execute(); fail("Should have thrown"); } catch (Exception e) { InOrder inOrder = Mockito.inOrder(o1, o2, o3); inOrder.verify(o3).start(); inOrder.verify(o2).start(); Mockito.verify(o1).abort(); Mockito.verify(o2).abort(); Mockito.verify(o3).abort(); Mockito.verifyNoMoreInteractions(o1, o2, o3); } }
@Test public void testExceptionInFinishAbortsAllOperations() throws Exception { Operation o1 = Mockito.mock(Operation.class); Operation o2 = Mockito.mock(Operation.class); Operation o3 = Mockito.mock(Operation.class); Mockito.doThrow(new Exception("in finish")).when(o2).finish(); ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet( Arrays.<Operation>asList(o1, o2, o3), counterSet, stateTracker)) { executor.execute(); fail("Should have thrown"); } catch (Exception e) { InOrder inOrder = Mockito.inOrder(o1, o2, o3); inOrder.verify(o3).start(); inOrder.verify(o2).start(); inOrder.verify(o1).start(); inOrder.verify(o1).finish(); inOrder.verify(o2).finish(); Mockito.verify(o1).abort(); Mockito.verify(o2).abort(); Mockito.verify(o3).abort(); Mockito.verifyNoMoreInteractions(o1, o2, o3); } }
@Test public void testExceptionInAbortSuppressed() throws Exception { Operation o1 = Mockito.mock(Operation.class); Operation o2 = Mockito.mock(Operation.class); Operation o3 = Mockito.mock(Operation.class); Operation o4 = Mockito.mock(Operation.class); Mockito.doThrow(new Exception("in finish")).when(o2).finish(); Mockito.doThrow(new Exception("suppressed in abort")).when(o3).abort(); ExecutionStateTracker stateTracker = ExecutionStateTracker.newForTest(); try (IntrinsicMapTaskExecutor executor = IntrinsicMapTaskExecutor.withSharedCounterSet( Arrays.<Operation>asList(o1, o2, o3, o4), counterSet, stateTracker)) { executor.execute(); fail("Should have thrown"); } catch (Exception e) { InOrder inOrder = Mockito.inOrder(o1, o2, o3, o4); inOrder.verify(o4).start(); inOrder.verify(o3).start(); inOrder.verify(o2).start(); inOrder.verify(o1).start(); inOrder.verify(o1).finish(); inOrder.verify(o2).finish(); Mockito.verify(o1).abort(); Mockito.verify(o2).abort(); Mockito.verify(o3).abort(); Mockito.verify(o4).abort(); Mockito.verifyNoMoreInteractions(o1, o2, o3, o4); assertThat(e.getMessage(), equalTo("in finish")); assertThat(e.getSuppressed(), arrayWithSize(1)); assertThat(e.getSuppressed()[0].getMessage(), equalTo("suppressed in abort")); } }