@Nonnull
    public static <T> BatchSource<T> batchFromProcessor(
            @Nonnull String sourceName,
            @Nonnull ProcessorMetaSupplier metaSupplier
    ) {
        checkSerializable(metaSupplier, "metaSupplier");
        return new BatchSourceTransform<>(sourceName, metaSupplier);
    }
@Test
    public void fromProcessor() {
        // Given
        List<Integer> input = sequence(itemCount);
        putToBatchSrcMap(input);

        // When
        BatchSource<Integer> source = Sources.batchFromProcessor("test",
                readMapP(srcName, truePredicate(), Entry::getValue));

        // Then
        p.readFrom(source).writeTo(sink);
        execute();
        assertEquals(toBag(input), sinkToBag());
    }