WriteFileP implements Processor { 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); } private  WriteFileP(
            @Nonnull String directoryName,
            @Nonnull FunctionEx<? super T, ? extends String> toStringFn,
            @Nonnull String charset,
            @Nullable String dateFormatter,
            long maxFileSize,
            boolean exactlyOnce,
            @Nonnull LongSupplier clock
    ); @Override void init(@Nonnull Outbox outbox, @Nonnull Context context); @Override boolean tryProcess(); @Override void process(int ordinal, @Nonnull Inbox inbox); @Override boolean tryProcessWatermark(@Nonnull Watermark watermark); @Override boolean complete(); @Override void close(); @Override boolean snapshotCommitPrepare(); @Override boolean snapshotCommitFinish(boolean success); @Override void restoreFromSnapshot(@Nonnull Inbox inbox); static ProcessorMetaSupplier metaSupplier(
            @Nonnull String directoryName,
            @Nonnull FunctionEx<? super T, ? extends String> toStringFn,
            @Nonnull String charset,
            @Nullable String datePattern,
            long maxFileSize,
            boolean exactlyOnce
    );  }
@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().newJob(dag); for (int i = 0; i < numItems; i++) { semaphore.release(); String stringValue = i + System.lineSeparator(); Path file = directory.resolve(String.format("%03d-0", i)); assertTrueEventually(() -> assertTrue("file not found: " + file, Files.exists(file)), 5); assertTrueEventually(() -> assertEquals(stringValue, new String(Files.readAllBytes(file), StandardCharsets.UTF_8)), 5); clock.incrementAndGet(); } job.join(); }
@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)); 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().newJob(dag); 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(), new String(Files.readAllBytes(file))); } job.join(); }