public static <T> ProcessorMetaSupplier metaSupplier(
            @Nonnull String directoryName,
            @Nonnull FunctionEx<? super T, ? extends String> toStringFn,
            @Nonnull String charset,
            @Nullable String datePattern,
            long maxFileSize,
            boolean exactlyOnce
    ) {
        return metaSupplier(directoryName, toStringFn, charset, datePattern, maxFileSize, exactlyOnce, SYSTEM_CLOCK);
    }
@Test
    public void test_rollByFileSize() throws Exception {
        int numItems = 10;
        DAG dag = new DAG();
        Vertex src = dag.newVertex("src", () -> new SlowSourceP(semaphore, numItems)).localParallelism(1);
        Vertex map = dag.newVertex("map", mapP((Integer i) -> i + 100));
        // maxFileSize is always large enough for 1 item but never for 2, both with windows and linux newlines
        long maxFileSize = 6L;
        Vertex sink = dag.newVertex("sink", WriteFileP.metaSupplier(
                directory.toString(), Objects::toString, "utf-8", null, maxFileSize, true));
        dag.edge(between(src, map));
        dag.edge(between(map, sink));

        Job job = instance().getJet().newJob(dag);

        // Then
        for (int i = 0; i < numItems; i++) {
            semaphore.release();
            int finalI = i;
            assertTrueEventually(() -> checkFileContents(100, finalI + 101, false, false, true));
        }
        for (int i = 0, j = 100; i < numItems / 2; i++) {
            Path file = directory.resolve("0-" + i);
            assertEquals((j++) + System.lineSeparator() + (j++) + System.lineSeparator(),
                    Files.readString(file));
        }

        job.join();
    }