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 testThreadSafeTimestampObservingWatermarkEstimator() throws Exception {
    WatermarkEstimators.WatermarkAndStateObserver<Instant> threadsafeWatermarkEstimator =
        WatermarkEstimators.threadSafe(
            new org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators
                .MonotonicallyIncreasing(GlobalWindow.TIMESTAMP_MIN_VALUE));
    testWatermarkEstimatorSnapshotsStateWithCompetingThread(
        threadsafeWatermarkEstimator,
        ((TimestampObservingWatermarkEstimator) threadsafeWatermarkEstimator)::observeTimestamp);
  }