@Override
  public CompletionStage<Collection<WindowedValue<OutT>>> finish() {
    /*
     * We can ignore the results here because its okay to call finish without invoking prepare. It will be a no-op
     * and an empty collection will be returned.
     */
    collectorSealed.compareAndSet(false, true);

    synchronized (this) {
      final CompletionStage<Collection<WindowedValue<OutT>>> sealedOutputFuture = outputFuture;
      outputFuture = CompletableFuture.completedFuture(new ArrayList<>());
      return sealedOutputFuture;
    }
  }
@Test
  public void testFinishWithoutPrepareReturnsEmptyCollection() {
    CompletionStage<Collection<WindowedValue<String>>> resultFuture = futureCollector.finish();
    CompletionStage<Void> validationFuture =
        resultFuture.thenAccept(
            result -> {
              Assert.assertTrue("Expected the result to be empty", result.isEmpty());
            });
    validationFuture.toCompletableFuture().join();
  }