public static <UserT, DestinationT, OutputT> WriteFiles<UserT, DestinationT, OutputT> to(
      FileBasedSink<UserT, DestinationT, OutputT> sink) {
    checkArgument(sink != null, "sink can not be null");
    return new AutoValue_WriteFiles.Builder<UserT, DestinationT, OutputT>()
        .setSink(sink)
        .setComputeNumShards(null)
        .setNumShardsProvider(null)
        .setWindowedWrites(false)
        .setWithAutoSharding(false)
        .setMaxNumWritersPerBundle(DEFAULT_MAX_NUM_WRITERS_PER_BUNDLE)
        .setSideInputs(sink.getDynamicDestinations().getSideInputs())
        .setSkipIfEmpty(false)
        .setBadRecordErrorHandler(new DefaultErrorHandler<>())
        .setBadRecordRouter(BadRecordRouter.THROWING_ROUTER)
        .build();
  }
@Test
  @Category(NeedsRunner.class)
  public void testWriteWindowed() throws IOException {
    List<String> inputs =
        Arrays.asList(
            "Critical canary",
            "Apprehensive eagle",
            "Intimidating pigeon",
            "Pedantic gull",
            "Frisky finch");
    runWrite(
        inputs,
        new WindowAndReshuffle<>(Window.into(FixedWindows.of(Duration.millis(2)))),
        getBaseOutputFilename(),
        WriteFiles.to(makeSimpleSink()));
  }