@Nonnull
    public static <T extends Entry> Sink<T> remoteCache(
            @Nonnull String cacheName,
            @Nonnull ClientConfig clientConfig
    ) {
        return fromProcessor("remoteCacheSink(" + cacheName + ')', writeRemoteCacheP(cacheName, clientConfig));
    }
@Test
    public void remoteCache() {
        // Given
        List<Integer> input = sequence(itemCount);
        putToBatchSrcCache(input);

        // When
        Sink<Entry<String, Integer>> sink = Sinks.remoteCache(sinkName, clientConfig);

        // Then
        p.readFrom(Sources.<String, Integer>cache(srcName)).writeTo(sink);
        execute();
        List<Entry<String, Integer>> expected = input.stream()
                                                     .map(i -> entry(String.valueOf(i), i))
                                                     .collect(toList());
        ICache<String, Integer> remoteCache = remoteHz.getCacheManager().getCache(sinkName);
        assertEquals(expected.size(), remoteCache.size());
        expected.forEach(entry -> assertEquals(entry.getValue(), remoteCache.get(entry.getKey())));
    }