@Nonnull
    public static <K, V> Sink<ChangeRecord> remoteMap(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends K> keyFn,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends V> valueFn
    ) {
        String name = "remoteMapCdcSink(" + mapName + ')';
        return sink(name, mapName, clientConfig, keyFn, valueFn);
    }
@Test
    public void deleteFromRemoteMap_ViaValueProjection() {
        remoteCluster = createRemoteCluster(smallInstanceConfig().setClusterName(randomName()), 2);
        HazelcastInstance remoteInstance = remoteCluster.get(0);

        ClientConfig clientConfig = getClientConfigForRemoteCluster(remoteInstance);

        p.readFrom(items(SYNC1, INSERT2))
                .writeTo(remoteSync(clientConfig));
        execute().join();

        p = Pipeline.create();
        p.readFrom(items(UPDATE1))
                .writeTo(CdcSinks.remoteMap(MAP, clientConfig,
                        r -> (Integer) r.key().toMap().get(ID),
                        r -> null
                ));
        execute().join();

        assertMap(remoteInstance, null, "gbailey@foobar.com");

        remoteInstance.getMap(MAP).destroy();
    }