@Nonnull
    public static <C> SourceBuilder<C>.Batch<Void> batch(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    ) {
        return new SourceBuilder<C>(name, createFn).new Batch<>();
    }
@Test
    public void batch_fileSource() throws Exception {
        // Given
        File textFile = createTestFile();

        // When
        BatchSource<String> fileSource = SourceBuilder
                .batch("file-source", x -> fileReader(textFile))
                .<String>fillBufferFn((in, buf) -> {
                    String line = in.readLine();
                    if (line != null) {
                        buf.add(line);
                    } else {
                        buf.close();
                    }
                })
                .destroyFn(BufferedReader::close)
                .build();

        // Then
        Pipeline p = Pipeline.create();
        p.readFrom(fileSource)
                .writeTo(sinkList());

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

        assertEquals(
                IntStream.range(0, itemCount).mapToObj(i -> "line" + i).collect(toList()),
                new ArrayList<>(sinkList)
        );
    }