public static IntrinsicMapTaskExecutor withSharedCounterSet(
      List<Operation> operations,
      CounterSet counters,
      ExecutionStateTracker executionStateTracker) {
    return new IntrinsicMapTaskExecutor(operations, counters, executionStateTracker);
  }
@Test
  @SuppressWarnings("unchecked")
  /**
   * This test makes sure that any metrics reported within an operation are part of the metric
   * containers returned by {@link getMetricContainers}.
   */
  public void testGetMetricContainers() throws Exception {
    ExecutionStateTracker stateTracker =
        new DataflowExecutionStateTracker(
            ExecutionStateSampler.newForTest(),
            new TestDataflowExecutionState(
                NameContext.forStage("testStage"),
                "other",
                null /* requestingStepName */,
                null /* sideInputIndex */,
                null /* metricsContainer */,
                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)) {
      // Call execute so that we run all the counters
      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)));
    }
  }