@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 stream_socketSource_withTimestamps_andLateness() throws IOException {
        // Given
        try (ServerSocket serverSocket = new ServerSocket(0)) {
            startServer(serverSocket);

            // When
            int localPort = serverSocket.getLocalPort();
            FunctionEx<String, Long> timestampFn = line -> Long.valueOf(line.substring(LINE_PREFIX.length()));

            int lateness = 10;
            long lastExpectedTs = itemCount - lateness;
            StreamSource<String> socketSource = SourceBuilder
                    .timestampedStream("socket-source-with-timestamps", ctx -> socketReader(localPort))
                    .<String>fillBufferFn((in, buf) -> {
                        String line = in.readLine();
                        if (line != null) {
                            long ts = timestampFn.apply(line);
                            buf.add(line, ts);
                            if (ts >= lastExpectedTs) {
                                System.out.println(line);
                            }
                        }
                    })
                    .destroyFn(BufferedReader::close)
                    .build();

            // Then
            Pipeline p = Pipeline.create();
            p.readFrom(socketSource)
                    .withNativeTimestamps(lateness)
                    .window(tumbling(1))
                    .aggregate(AggregateOperations.counting())
                    .writeTo(sinkList());

            hz().getJet().newJob(p);

            List<WindowResult<Long>> expected = LongStream.range(1, itemCount - lateness)
                    .mapToObj(i -> new WindowResult<>(i - 1, i, 1L))
                    .collect(toList());

            assertTrueEventually(() -> assertEquals(expected, new ArrayList<>(sinkList)), 10);
        }
    }