public static <WatermarkEstimatorStateT>
      WatermarkAndStateObserver<WatermarkEstimatorStateT> threadSafe(
          WatermarkEstimator<WatermarkEstimatorStateT> watermarkEstimator) {
    if (watermarkEstimator instanceof TimestampObservingWatermarkEstimator) {
      return new ThreadSafeTimestampObservingWatermarkEstimator<>(watermarkEstimator);
    } else if (watermarkEstimator instanceof ManualWatermarkEstimator) {
      return new ThreadSafeManualWatermarkEstimator<>(watermarkEstimator);
    }
    return new ThreadSafeWatermarkEstimator<>(watermarkEstimator);
  }
@Test
  public void testThreadSafeWatermarkEstimator() throws Exception {
    Instant[] reference = new Instant[] {GlobalWindow.TIMESTAMP_MIN_VALUE};
    WatermarkEstimator<Instant> watermarkEstimator =
        new WatermarkEstimator<Instant>() {
          private Instant currentWatermark = GlobalWindow.TIMESTAMP_MIN_VALUE;

          @Override
          public Instant currentWatermark() {
            currentWatermark = reference[0];
            return currentWatermark;
          }

          @Override
          public Instant getState() {
            return currentWatermark;
          }
        };
    testWatermarkEstimatorSnapshotsStateWithCompetingThread(
        WatermarkEstimators.threadSafe(watermarkEstimator), (instant) -> reference[0] = instant);
  }