public WriteFileP(
            @Nonnull String directoryName,
            @Nonnull FunctionEx<? super T, ? extends String> toStringFn,
            @Nonnull String charset,
            @Nullable String dateFormatter,
            long maxFileSize,
            boolean exactlyOnce,
            @Nonnull LongSupplier clock
    ) {
        this.directory = Paths.get(directoryName);
        this.toStringFn = toStringFn;
        this.charset = Charset.forName(charset);
        this.dateFormatter = dateFormatter != null
                ? DateTimeFormatter.ofPattern(dateFormatter).withZone(ZoneId.from(ZoneOffset.UTC))
                : null;
        this.maxFileSize = maxFileSize;
        this.exactlyOnce = exactlyOnce;
        this.clock = clock;
    }
@Test
    public void when_slowSource_then_fileFlushedAfterEachItem() {
        // Given
        int numItems = 10;

        DAG dag = new DAG();
        Vertex source = dag.newVertex("source", () -> new SlowSourceP(semaphore, numItems))
                           .localParallelism(1);
        Vertex sink = dag.newVertex("sink",
                writeFileP(directory.toString(), StandardCharsets.UTF_8, null, DISABLE_ROLLING, true, Object::toString))
                         .localParallelism(1);
        dag.edge(between(source, sink));

        Job job = instance().getJet().newJob(dag);
        for (int i = 0; i < numItems; i++) {
            // When
            semaphore.release();
            int finalI = i;
            // Then
            assertTrueEventually(() -> checkFileContents(0, finalI + 1, false, false, true), 5);
        }

        // wait for the job to finish
        job.join();
    }