@Override
  public Instant currentInputWatermarkTime() {
    return watermarks.getInputWatermark();
  }
@Test
  public void getInputWatermarkTimeUsesWatermarkTime() {
    when(watermarks.getInputWatermark()).thenReturn(new Instant(8765L));
    assertThat(internals.currentInputWatermarkTime(), equalTo(new Instant(8765L)));
  }