@Nonnull
    public static <E, K, V, R> Sink<E> mapWithEntryProcessor(
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super E, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super E, ? extends EntryProcessor<K, V, R>> toEntryProcessorFn
    ) {
        return Sinks.<E, K, V, R>mapEntryProcessorBuilder(mapName)
                .toKeyFn(toKeyFn)
                .toEntryProcessorFn(toEntryProcessorFn)
                .build();
    }
@Test
    @Category(SlowTest.class)
    public void mapWithEntryProcessor_testBackpressure() {
        /*
        NOTE TO THE TEST
        This test tries to test that when a permit for async op is denied, the
        processor correctly yields. We don't assert that it actually happened,
        nor that the backpressure was applied to the upstream. We only try to
        simulate a slow sink (using the SleepingEntryProcessor) and check that
        the results are correct.
         */
        String targetMap = randomMapName();

        List<Integer> input = sequence(5_001);
        p.readFrom(TestSources.items(input))
         .writeTo(Sinks.mapWithEntryProcessor(targetMap, FunctionEx.identity(), SleepingEntryProcessor::new));
        execute();
        Map<Integer, Integer> actual = new HashMap<>(hz().getMap(targetMap));
        Map<Integer, Integer> expected =
                input.stream()
                     .collect(toMap(Function.identity(), Function.identity(), Integer::sum));
        assertEquals(expected, actual);
    }