@Nonnull
    public static <K, V> Sink<Entry<K, V>> map(@Nonnull String mapName) {
        return map(mapName, Entry::getKey, Entry::getValue);
    }
@Test
    @SuppressWarnings("unchecked")
    public void test_adaptingPartitionFunction() {
        Pipeline p = Pipeline.create();
        StreamStage<KeyedWindowResult<String, Long>> input1 =
                p.readFrom(TestSources.items(0))
                 .addTimestamps(i -> i, 0)
                 .groupingKey(item -> "key0")
                 .window(WindowDefinition.sliding(1, 1))
                 .aggregate(AggregateOperations.counting());
        BatchStage<Entry<String, Long>> input2 =
                p.readFrom(TestSources.items(entry("key1", 2L)));

        IMap<String, Long> sinkMap = hz().getMap(randomMapName());
        p.writeTo(Sinks.map(sinkMap), input1, input2);

        hz().getJet().newJob(p).join();
        assertEquals(2, sinkMap.size());
        assertEquals((Long) 1L, sinkMap.get("key0"));
        assertEquals((Long) 2L, sinkMap.get("key1"));
    }