@Nonnull
    public static <T, K, V> StreamSource<T> mapJournal(
            @Nonnull String mapName,
            @Nonnull JournalInitialPosition initialPos,
            @Nonnull FunctionEx<? super EventJournalMapEvent<K, V>, ? extends T> projectionFn,
            @Nonnull PredicateEx<? super EventJournalMapEvent<K, V>> predicateFn
    ) {
        return streamFromProcessorWithWatermarks("mapJournalSource(" + mapName + ')',
                false, w -> streamMapP(mapName, predicateFn, projectionFn, initialPos, w));
    }
@Test(expected = IllegalStateException.class)
    public void when_streamSourceUsedTwice_then_throwException() {
        // Given
        StreamSource<Entry<Object, Object>> source = Sources.mapJournal(srcName, START_FROM_CURRENT);
        p.readFrom(source);

        // When-Then
        p.readFrom(source);
    }