LateDataUtils { public static Instant garbageCollectionTime( BoundedWindow window, WindowingStrategy windowingStrategy) { return garbageCollectionTime(window, windowingStrategy.getAllowedLateness()); } private  LateDataUtils(); static Instant garbageCollectionTime(
      BoundedWindow window, WindowingStrategy windowingStrategy); static Instant garbageCollectionTime(BoundedWindow window, Duration allowedLateness); static Iterable<WindowedValue<V>> dropExpiredWindows(
      final K key,
      Iterable<WindowedValue<V>> elements,
      final TimerInternals timerInternals,
      final WindowingStrategy<?, ?> windowingStrategy,
      final CounterCell droppedDueToLateness);  }
@Test public void beforeEndOfGlobalWindowSame() { FixedWindows windowFn = FixedWindows.of(Duration.standardMinutes(5)); Duration allowedLateness = Duration.standardMinutes(2); WindowingStrategy<?, ?> strategy = WindowingStrategy.globalDefault() .withWindowFn(windowFn) .withAllowedLateness(allowedLateness); IntervalWindow window = windowFn.assignWindow(new Instant(10)); assertThat( LateDataUtils.garbageCollectionTime(window, strategy), equalTo(window.maxTimestamp().plus(allowedLateness))); }
@Test public void garbageCollectionTimeAfterEndOfGlobalWindow() { FixedWindows windowFn = FixedWindows.of(Duration.standardMinutes(5)); WindowingStrategy<?, ?> strategy = WindowingStrategy.globalDefault().withWindowFn(windowFn); IntervalWindow window = windowFn.assignWindow(new Instant(BoundedWindow.TIMESTAMP_MAX_VALUE)); assertThat(window.maxTimestamp(), equalTo(GlobalWindow.INSTANCE.maxTimestamp())); assertThat( LateDataUtils.garbageCollectionTime(window, strategy), equalTo(GlobalWindow.INSTANCE.maxTimestamp())); }
@Test public void garbageCollectionTimeAfterEndOfGlobalWindowWithLateness() { FixedWindows windowFn = FixedWindows.of(Duration.standardMinutes(5)); Duration allowedLateness = Duration.millis(Long.MAX_VALUE); WindowingStrategy<?, ?> strategy = WindowingStrategy.globalDefault() .withWindowFn(windowFn) .withAllowedLateness(allowedLateness); IntervalWindow window = windowFn.assignWindow(new Instant(-100)); assertThat( window.maxTimestamp().plus(allowedLateness), Matchers.greaterThan(GlobalWindow.INSTANCE.maxTimestamp())); assertThat( LateDataUtils.garbageCollectionTime(window, strategy), equalTo(GlobalWindow.INSTANCE.maxTimestamp())); }