BatchModeExecutionContext extends DataflowExecutionContext<BatchModeExecutionContext.StepContext> { public Iterable<CounterUpdate> extractMsecCounters(boolean isFinalUpdate) { return executionStateRegistry.extractUpdates(isFinalUpdate); } private  BatchModeExecutionContext(
      CounterFactory counterFactory,
      Cache<?, WeightedValue<?>> dataCache,
      Cache<?, ?> logicalReferenceCache,
      ReaderFactory readerFactory,
      PipelineOptions options,
      DataflowExecutionStateTracker executionStateTracker,
      DataflowExecutionStateRegistry executionStateRegistry); static BatchModeExecutionContext forTesting(
      PipelineOptions options, CounterFactory counterFactory, String stageName); static BatchModeExecutionContext forTesting(PipelineOptions options, String stageName); static BatchModeExecutionContext create(
      CounterFactory counterFactory,
      Cache<?, WeightedValue<?>> dataCache,
      Cache<?, ?> logicalReferenceCache,
      ReaderFactory readerFactory,
      DataflowPipelineOptions options,
      String stageName,
      String workItemId); void setKey(Object key); @Nullable Object getKey(); Cache<K, V> getDataCache(); Cache<K, V> getLogicalReferenceCache(); Iterable<CounterUpdate> extractMetricUpdates(boolean isFinalUpdate); void commitMetricUpdates(); Iterable<CounterUpdate> extractMsecCounters(boolean isFinalUpdate); Long extractThrottleTime();  }
@Test public void extractMsecCounters() { BatchModeExecutionContext executionContext = BatchModeExecutionContext.forTesting(PipelineOptionsFactory.create(), "testStage"); MetricsContainer metricsContainer = Mockito.mock(MetricsContainer.class); ProfileScope otherScope = Mockito.mock(ProfileScope.class); ProfileScope profileScope = Mockito.mock(ProfileScope.class); ExecutionState start1 = executionContext.executionStateRegistry.getState( NameContext.create("stage", "original-1", "system-1", "user-1"), ExecutionStateTracker.START_STATE_NAME, metricsContainer, profileScope); ExecutionState process1 = executionContext.executionStateRegistry.getState( NameContext.create("stage", "original-1", "system-1", "user-1"), ExecutionStateTracker.PROCESS_STATE_NAME, metricsContainer, profileScope); ExecutionState start2 = executionContext.executionStateRegistry.getState( NameContext.create("stage", "original-2", "system-2", "user-2"), ExecutionStateTracker.START_STATE_NAME, metricsContainer, profileScope); ExecutionState other = executionContext.executionStateRegistry.getState( NameContext.forStage("stage"), "other", null, NoopProfileScope.NOOP); other.takeSample(120); start1.takeSample(100); process1.takeSample(500); assertThat( executionContext.extractMsecCounters(false), containsInAnyOrder( msecStage("other-msecs", "stage", 120), msec("start-msecs", "stage", "original-1", 100), msec("process-msecs", "stage", "original-1", 500))); process1.takeSample(200); start2.takeSample(200); assertThat( executionContext.extractMsecCounters(false), containsInAnyOrder( msec("process-msecs", "stage", "original-1", 500 + 200), msec("start-msecs", "stage", "original-2", 200))); process1.takeSample(300); assertThat( executionContext.extractMsecCounters(true), hasItems( msecStage("other-msecs", "stage", 120), msec("start-msecs", "stage", "original-1", 100), msec("process-msecs", "stage", "original-1", 500 + 200 + 300), msec("start-msecs", "stage", "original-2", 200))); }