@Override
  public void processWatermark(Instant watermark, OpEmitter<OutT> emitter) {
    // propagate watermark immediately if no bundle is in progress and all the previous bundles have
    // completed.
    if (shouldProcessWatermark()) {
      LOG.debug("Propagating watermark: {} directly since no bundle in progress.", watermark);
      bundleProgressListener.onWatermark(watermark, emitter);
      return;
    }

    // hold back the watermark since there is either a bundle in progress or previously closed
    // bundles are unfinished.
    this.bundleWatermarkHold = watermark;
  }
@Test
  public void testProcessWatermarkWhenBundleNotStarted() {
    Instant watermark = new Instant();
    portableBundleManager =
        new PortableBundleManager<>(bundleProgressListener, 4, 1, bundleTimerScheduler, TIMER_ID);
    portableBundleManager.processWatermark(watermark, emitter);
    verify(bundleProgressListener, times(1)).onWatermark(eq(watermark), eq(emitter));
  }