SourceBuilder { @Nonnull public static <C> SourceBuilder<C>.Stream<Void> stream( @Nonnull String name, @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn ) { return new SourceBuilder<C>(name, createFn).new Stream<Void>(); } private  SourceBuilder(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Context, ? extends C> createFn
    ); @Nonnull static SourceBuilder<C>.Batch<Void> batch(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    ); @Nonnull static SourceBuilder<C>.Stream<Void> stream(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    ); @Nonnull static SourceBuilder<C>.TimestampedStream<Void> timestampedStream(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    );  }
@Test public void test_nonFaultTolerantSource_processingGuaranteeOn() { StreamSource<Integer> source = SourceBuilder .stream("src", procCtx -> "foo") .<Integer>fillBufferFn((ctx, buffer) -> { buffer.add(0); Thread.sleep(100); }) .build(); Pipeline p = Pipeline.create(); IList<Integer> result = jet().getList("result-" + UuidUtil.newUnsecureUuidString()); p.readFrom(source) .withoutTimestamps() .writeTo(Sinks.list(result)); Job job = jet().newJob(p, new JobConfig().setProcessingGuarantee(EXACTLY_ONCE).setSnapshotIntervalMillis(100)); JobRepository jr = new JobRepository(jet()); waitForFirstSnapshot(jr, job.getId(), 10, true); job.restart(); assertJobStatusEventually(job, JobStatus.RUNNING); int currentSize = result.size(); assertTrueEventually(() -> assertTrue(result.size() > currentSize), 5); }