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_rollByDate() {
        int numItems = 10;
        DAG dag = new DAG();
        Vertex src = dag.newVertex("src", () -> new SlowSourceP(semaphore, numItems)).localParallelism(1);
        @SuppressWarnings("Convert2MethodRef")
        Vertex sink = dag.newVertex("sink", WriteFileP.metaSupplier(
                directory.toString(), Objects::toString, "utf-8", "SSS", DISABLE_ROLLING, true,
                (LongSupplier & Serializable) () -> clock.get()));
        dag.edge(between(src, sink));

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

        for (int i = 0; i < numItems; i++) {
            // When
            semaphore.release();
            String stringValue = i + System.lineSeparator();
            // Then
            Path file = directory.resolve(String.format("%03d-0", i));
            assertTrueEventually(() -> assertTrue("file not found: " + file, Files.exists(file)), 5);
            assertTrueEventually(() ->
                    assertEquals(stringValue, Files.readString(file)), 5);
            clock.incrementAndGet();
        }

        job.join();
    }