@Override
  public void add(CompletionStage<WindowedValue<OutT>> element) {
    checkState(
        !collectorSealed.get(),
        "Cannot add element to an unprepared collector. Make sure prepare() is invoked before adding elements.");

    // We need synchronize guard against scenarios when watermark/finish bundle trigger outputs.
    synchronized (this) {
      outputFuture =
          outputFuture.thenCombine(
              element,
              (collection, event) -> {
                collection.add(event);
                return collection;
              });
    }
  }
@Test(expected = IllegalStateException.class)
  public void testAddWithoutPrepareCallThrowsException() {
    futureCollector.add(mock(CompletionStage.class));
  }