@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_distributed() throws Exception {
        // Given
        File textFile = createTestFile();

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

        // Then
        Pipeline p = Pipeline.create();
        p.readFrom(fileSource)
                .writeTo(sinkList());
        hz().getJet().newJob(p).join();

        Map<String, Integer> actual = sinkToBag();
        Map<String, Integer> expected = IntStream.range(0, itemCount)
                .boxed()
                .collect(Collectors.toMap(i -> "line" + i, i -> PREFERRED_LOCAL_PARALLELISM * MEMBER_COUNT));

        assertEquals(expected, actual);
    }