public ExecutionStateTracker create() {
    return new ExecutionStateTracker();
  }
@Test
  public void testCountersReturnedAreBasedUponCurrentExecutionState() throws Exception {
    MillisProvider clock = mock(MillisProvider.class);
    ExecutionStateSampler sampler =
        new ExecutionStateSampler(
            PipelineOptionsFactory.fromArgs("--experiments=state_sampling_period_millis=10")
                .create(),
            clock);
    ExecutionStateTracker tracker = sampler.create();
    MetricsEnvironment.setCurrentContainer(tracker.getMetricsContainer());
    ExecutionState state = tracker.create("shortId", "ptransformId", "uniqueName", "state");

    state.activate();
    TEST_USER_COUNTER.inc();
    TEST_USER_DISTRIBUTION.update(2);
    TEST_USER_GAUGE.set(3);
    TEST_USER_STRING_SET.add("ab");
    TEST_USER_HISTOGRAM.update(4);
    state.deactivate();

    TEST_USER_COUNTER.inc(11);
    TEST_USER_DISTRIBUTION.update(12);
    TEST_USER_GAUGE.set(13);
    TEST_USER_STRING_SET.add("cd");
    TEST_USER_HISTOGRAM.update(14);
    TEST_USER_HISTOGRAM.update(14);

    // Verify the execution state was updated
    assertEquals(
        1L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getContainer("ptransformId")
                .getCounter(TEST_USER_COUNTER.getName())
                .getCumulative());
    assertEquals(
        2L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getContainer("ptransformId")
                .getDistribution(TEST_USER_DISTRIBUTION.getName())
                .getCumulative()
                .sum());
    assertEquals(
        3L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getContainer("ptransformId")
                .getGauge(TEST_USER_GAUGE.getName())
                .getCumulative()
                .value());
    assertEquals(
        ImmutableSet.of("ab"),
        tracker
            .getMetricsContainerRegistry()
            .getContainer("ptransformId")
            .getStringSet(TEST_USER_STRING_SET.getName())
            .getCumulative()
            .stringSet());
    assertEquals(
        1L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getContainer("ptransformId")
                .getHistogram(
                    TEST_USER_HISTOGRAM.getName(), HistogramData.LinearBuckets.of(0, 100, 1))
                .getCumulative()
                .getCount(0));

    // Verify the unbound container
    assertEquals(
        11L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getUnboundContainer()
                .getCounter(TEST_USER_COUNTER.getName())
                .getCumulative());
    assertEquals(
        12L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getUnboundContainer()
                .getDistribution(TEST_USER_DISTRIBUTION.getName())
                .getCumulative()
                .sum());
    assertEquals(
        13L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getUnboundContainer()
                .getGauge(TEST_USER_GAUGE.getName())
                .getCumulative()
                .value());
    assertEquals(
        ImmutableSet.of("cd"),
        tracker
            .getMetricsContainerRegistry()
            .getUnboundContainer()
            .getStringSet(TEST_USER_STRING_SET.getName())
            .getCumulative()
            .stringSet());
    assertEquals(
        2L,
        (long)
            tracker
                .getMetricsContainerRegistry()
                .getUnboundContainer()
                .getHistogram(
                    TEST_USER_HISTOGRAM.getName(), HistogramData.LinearBuckets.of(0, 100, 1))
                .getCumulative()
                .getCount(0));
  }