CdcSinks { @Nonnull public static <K, V> Sink<ChangeRecord> map( @Nonnull String mapName, @Nonnull FunctionEx<? super ChangeRecord, ? extends K> keyFn, @Nonnull FunctionEx<? super ChangeRecord, ? extends V> valueFn ) { String name = "mapCdcSink(" + mapName + ')'; return sink(name, mapName, null, keyFn, valueFn); } private  CdcSinks(); @Nonnull static Sink<ChangeRecord> map(
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends K> keyFn,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends V> valueFn
    ); @Nonnull static Sink<ChangeRecord> map(
            @Nonnull IMap<? super K, ? super V> map,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends K> keyFn,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends V> valueFn
    ); @Nonnull static Sink<ChangeRecord> remoteMap(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends K> keyFn,
            @Nonnull FunctionEx<? super ChangeRecord, ? extends V> valueFn
    ); static final HazelcastProperty SEQUENCE_CACHE_EXPIRATION_SECONDS; }
@Test public void deleteFromLocalMap_ViaValueProjection() { p.readFrom(items(() -> Arrays.asList(SYNC1, INSERT2).iterator())) .writeTo(localSync()); execute().join(); p = Pipeline.create(); p.readFrom(items(() -> Collections.singletonList(UPDATE1).iterator())) .writeTo(CdcSinks.map(MAP, r -> (Integer) r.key().toMap().get(ID), r -> null )); execute().join(); assertMap(jet(), null, "gbailey@foobar.com"); jet().getMap(MAP).destroy(); }
@Test public void reordering() { SupplierEx<Iterator<? extends ChangeRecord>> supplier = () -> Arrays.asList( SYNC1, UPDATE1, new ChangeRecordImpl(0, 10, UPDATE1.key().toJson(), UPDATE1.value().toJson().replace("sthomas@acme.com", "sthomas2@acme.com")), new ChangeRecordImpl(0, 11, UPDATE1.key().toJson(), UPDATE1.value().toJson().replace("sthomas@acme.com", "sthomas3@acme.com")), new ChangeRecordImpl(0, 12, UPDATE1.key().toJson(), UPDATE1.value().toJson().replace("sthomas@acme.com", "sthomas4@acme.com")), new ChangeRecordImpl(0, 13, UPDATE1.key().toJson(), UPDATE1.value().toJson().replace("sthomas@acme.com", "sthomas5@acme.com")) ).iterator(); Util.checkSerializable(supplier, "kaka"); p.readFrom(items(supplier)) .rebalance() .map(r -> r) .writeTo(localSync()); execute().join(); assertMap(jet(), "sthomas5@acme.com", null); jet().getMap(MAP).destroy(); }