@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 (!isBundleStarted() && pendingBundleCount.get() == 0) {
      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;

    // for batch mode, the max watermark should force the bundle to close
    if (BoundedWindow.TIMESTAMP_MAX_VALUE.equals(watermark)) {
      /*
       * Due to lack of async watermark function, we block on the previous watermark futures before propagating the watermark
       * downstream. If a bundle is in progress tryFinishBundle() fill force the bundle to close and emit watermark.
       * If no bundle in progress, we progress watermark explicitly after the completion of previous watermark futures.
       */
      if (isBundleStarted()) {
        LOG.info(
            "Received max watermark. Triggering finish bundle before flushing the watermark downstream.");
        tryFinishBundle(emitter);
        watermarkFuture.toCompletableFuture().join();
      } else {
        LOG.info(
            "Received max watermark. Waiting for previous bundles to complete before flushing the watermark downstream.");
        watermarkFuture.toCompletableFuture().join();
        bundleProgressListener.onWatermark(watermark, emitter);
      }
    }
  }
@Test
  public void testProcessWatermarkWhenNoBundleInProgress() {
    Instant now = Instant.now();
    OpEmitter<String> mockEmitter = mock(OpEmitter.class);
    bundleManager.processWatermark(now, mockEmitter);
    verify(bundleProgressListener, times(1)).onWatermark(now, mockEmitter);
  }