@Nonnull
    public static <C> SourceBuilder<C>.TimestampedStream<Void> timestampedStream(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    ) {
        return new SourceBuilder<C>(name, createFn).new TimestampedStream<>();
    }
@Test
    public void test_faultTolerance_snapshotWithUserDefinedObject() {
        StreamSource<WrappedInt> source = SourceBuilder
                .timestampedStream("src", ctx -> new WrappedNumberGeneratorContext())
                .<WrappedInt>fillBufferFn((src, buffer) -> {
                    for (int i = 0; i < 100; i++) {
                        buffer.add(src.current, src.current.value);
                        src.current = new WrappedInt(src.current.value + 1);
                    }
                    Thread.sleep(100);
                })
                .createSnapshotFn(src -> {
                    System.out.println("Will save " + src.current.value + " to snapshot");
                    return src;
                })
                .restoreSnapshotFn((src, states) -> {
                    assert states.size() == 1;
                    src.restore(states.get(0));
                    System.out.println("Restored " + src.current.value + " from snapshot");
                })
                .build();

        testFaultTolerance(source);
    }