public ExecutionStateTracker create() {
    return new ExecutionStateTracker();
  }
@Test
  public void testTrackerReuse() 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", "ptransformIdName", "process");

    CountDownLatch waitTillActive = new CountDownLatch(1);
    CountDownLatch waitTillSecondStateActive = new CountDownLatch(1);
    CountDownLatch waitForSamples = new CountDownLatch(1);
    CountDownLatch waitForMoreSamples = new CountDownLatch(1);
    Thread testThread = Thread.currentThread();
    Mockito.when(clock.getMillis())
        .thenAnswer(
            new Answer<Long>() {
              private long currentTime;

              @Override
              public Long answer(InvocationOnMock invocation) throws Throwable {
                if (Thread.currentThread().equals(testThread)) {
                  return 0L;
                } else {
                  // Block the state sampling thread till the state is active
                  // and unblock the state transition once a certain number of samples
                  // have been taken.
                  if (currentTime < 1000L) {
                    waitTillActive.await();
                    currentTime += 100L;
                  } else if (currentTime < 1500L) {
                    waitForSamples.countDown();
                    waitTillSecondStateActive.await();
                    currentTime += 100L;
                  } else {
                    waitForMoreSamples.countDown();
                  }
                  return currentTime;
                }
              }
            });

    {
      tracker.start("bundleId1");
      state.activate();
      waitTillActive.countDown();
      waitForSamples.await();
      TEST_USER_COUNTER.inc();
      state.deactivate();
      Map<String, ByteString> finalResults = new HashMap<>();
      tracker.updateFinalMonitoringData(finalResults);
      assertThat(
          MonitoringInfoEncodings.decodeInt64Counter(finalResults.get("shortId")),
          // Because we are using lazySet, we aren't guaranteed to see the latest value.
          // The CountDownLatch ensures that we will see either the prior value or
          // the latest value.
          anyOf(equalTo(900L), equalTo(1000L)));
      assertEquals(
          1L,
          (long)
              tracker
                  .getMetricsContainerRegistry()
                  .getContainer("ptransformId")
                  .getCounter(TEST_USER_COUNTER.getName())
                  .getCumulative());
      tracker.reset();
    }

    {
      tracker.start("bundleId2");
      state.activate();
      waitTillSecondStateActive.countDown();
      waitForMoreSamples.await();
      TEST_USER_COUNTER.inc();
      state.deactivate();
      Map<String, ByteString> finalResults = new HashMap<>();
      tracker.updateFinalMonitoringData(finalResults);
      assertThat(
          MonitoringInfoEncodings.decodeInt64Counter(finalResults.get("shortId")),
          // Because we are using lazySet, we aren't guaranteed to see the latest value.
          // The CountDownLatch ensures that we will see either the prior value or
          // the latest value.
          anyOf(equalTo(400L), equalTo(500L)));
      assertEquals(
          1L,
          (long)
              tracker
                  .getMetricsContainerRegistry()
                  .getContainer("ptransformId")
                  .getCounter(TEST_USER_COUNTER.getName())
                  .getCumulative());
      tracker.reset();
    }

    expectedLogs.verifyNotLogged("Operation ongoing");
  }