WriteFileP implements Processor { 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 ) { this.directory = Paths.get(directoryName); this.toStringFn = toStringFn; this.charset = Charset.forName(charset); this.dateFormatter = dateFormatter != null ? DateTimeFormatter.ofPattern(dateFormatter) : null; this.maxFileSize = maxFileSize; this.exactlyOnce = exactlyOnce; this.clock = 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 when_slowSource_then_fileFlushedAfterEachItem() { 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().newJob(dag); for (int i = 0; i < numItems; i++) { semaphore.release(); int finalI = i; assertTrueEventually(() -> checkFileContents(0, finalI + 1, false, false, true), 5); } job.join(); }
@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().newJob(dag, config); assertJobStatusEventually(job, 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); }