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 stressTest_snapshots_noRestarts() throws Exception {
        DAG dag = new DAG();
        int numItems = 5;
        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));

        JobConfig config = new JobConfig()
                .setProcessingGuarantee(EXACTLY_ONCE)
                .setSnapshotIntervalMillis(500);
        Job job = instance().getJet().newJob(dag, config);
        assertThat(job).eventuallyHasStatus(RUNNING);

        JobRepository jr = new JobRepository(instance());
        waitForFirstSnapshot(jr, job.getId(), 10, true);
        for (int i = 0; i < numItems; i++) {
            waitForNextSnapshot(jr, job.getId(), 10, true);
            semaphore.release();
        }
        job.join();

        checkFileContents(0, numItems, false, false, true);
    }