@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<>();
    }
@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 = hz().getList("result-" + UuidUtil.newUnsecureUuidString());
        p.readFrom(source)
         .withoutTimestamps()
         .writeTo(Sinks.list(result));

        Job job = hz().getJet().newJob(p, new JobConfig().setProcessingGuarantee(EXACTLY_ONCE).setSnapshotIntervalMillis(100));
        JobRepository jr = new JobRepository(hz());
        waitForFirstSnapshot(jr, job.getId(), 10, true);

        job.restart();
        assertThat(job).eventuallyHasStatus(JobStatus.RUNNING);
        int currentSize = result.size();
        assertTrueEventually(() -> assertTrue(result.size() > currentSize), 5);
    }