public void start() {
    if (timeLimit > 0 && this.flushFuture == null) {
      this.flushFuture =
          Executors.newSingleThreadScheduledExecutor(
                  new ThreadFactoryBuilder()
                      .setDaemon(true)
                      .setNameFormat("DataBufferOutboundFlusher-thread")
                      .build())
              .scheduleAtFixedRate(this::flush, timeLimit, timeLimit, TimeUnit.MILLISECONDS);
    }
  }
@Test
  public void testConfiguredTimeLimit() throws Exception {
    List<Elements> values = new ArrayList<>();
    PipelineOptions options = PipelineOptionsFactory.create();
    options
        .as(ExperimentalOptions.class)
        .setExperiments(Arrays.asList("data_buffer_time_limit_ms=1"));
    final CountDownLatch waitForFlush = new CountDownLatch(1);
    BeamFnDataOutboundAggregator aggregator =
        new BeamFnDataOutboundAggregator(
            options,
            endpoint::getInstructionId,
            TestStreams.withOnNext(
                    (Consumer<Elements>)
                        e -> {
                          values.add(e);
                          waitForFlush.countDown();
                        })
                .build(),
            false);

    // Test that it emits when time passed the time limit
    FnDataReceiver<byte[]> dataReceiver = registerOutputLocation(aggregator, endpoint, CODER);
    aggregator.start();
    dataReceiver.accept(new byte[1]);
    waitForFlush.await(); // wait the flush thread to flush the buffer
    Assert.assertEquals(messageWithData(new byte[1]), values.get(0));
  }