@Nonnull
    public static <K, V> BatchSource<Entry<K, V>> map(@Nonnull String mapName) {
        return batchFromProcessor("mapSource(" + mapName + ')', readMapP(mapName));
    }
@Test
    public void map_withProjectionToNull_then_nullsSkipped() {
        // given
        String mapName = randomName();
        IMap<Integer, Entry<Integer, String>> sourceMap = hz().getMap(mapName);
        range(0, itemCount).forEach(i -> sourceMap.put(i, entry(i, i % 2 == 0 ? null : String.valueOf(i))));

        // when
        BatchSource<String> source = Sources.map(mapName, truePredicate(), singleAttribute("value"));

        // then
        p.readFrom(source).writeTo(sink);
        hz().getJet().newJob(p);
        assertTrueEventually(() -> assertEquals(
                range(0, itemCount)
                        .filter(i -> i % 2 != 0)
                        .mapToObj(String::valueOf)
                        .sorted()
                        .collect(joining("\n")),
                hz().<String>getList(sinkName)
                        .stream()
                        .sorted()
                        .collect(joining("\n"))
        ));
    }