WatermarkEstimators { 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); }  static WatermarkAndStateObserver<WatermarkEstimatorStateT> threadSafe(
          WatermarkEstimator<WatermarkEstimatorStateT> 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); }
@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); }