@Override
  public @Nullable Instant currentOutputWatermarkTime() {
    return watermarks.getOutputWatermark();
  }
@Test
  public void getOutputWatermarkTimeUsesWatermarkTime() {
    when(watermarks.getOutputWatermark()).thenReturn(new Instant(25525L));
    assertThat(internals.currentOutputWatermarkTime(), equalTo(new Instant(25525L)));
  }