@Nonnull
    public static <K, V> BatchSource<Entry<K, V>> remoteCache(
            @Nonnull String cacheName,
            @Nonnull ClientConfig clientConfig
    ) {
        return batchFromProcessor("remoteCacheSource(" + cacheName + ')',
                ProcessorMetaSupplier.of(readRemoteCacheP(cacheName, clientConfig)));
    }
@Test
    public void remoteCache() {
        // Given
        List<Integer> input = sequence(itemCount);
        putToCache(remoteHz.getCacheManager().getCache(srcName), input);

        // When
        BatchSource<Entry<Object, Object>> source = Sources.remoteCache(srcName, clientConfig);

        // Then
        p.readFrom(source).writeTo(sink);
        execute();
        List<Entry<String, Integer>> expected = input.stream()
                                                     .map(i -> entry(String.valueOf(i), i))
                                                     .collect(toList());
        assertEquals(toBag(expected), sinkToBag());
    }