@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);
    }
@Test
    @SuppressWarnings({"rawtypes", "unchecked"})
    public void reordering() {
        SupplierEx[] records = {
                SYNC1,
                UPDATE1,
                () -> changeRecord(10, UPDATE, UPDATE1.get().key().toJson(), null,
                        UPDATE1.get().value().toJson().replace("sthomas@acme.com", "sthomas2@acme.com")),
                () -> changeRecord(11, UPDATE, UPDATE1.get().key().toJson(), null,
                        UPDATE1.get().value().toJson().replace("sthomas@acme.com", "sthomas3@acme.com")),
                () -> changeRecord(12, UPDATE, UPDATE1.get().key().toJson(), null,
                        UPDATE1.get().value().toJson().replace("sthomas@acme.com", "sthomas4@acme.com")),
                () -> changeRecord(13, UPDATE, UPDATE1.get().key().toJson(), null,
                        UPDATE1.get().value().toJson().replace("sthomas@acme.com", "sthomas5@acme.com"))
        };
        p.readFrom(items(records))
                .rebalance()
                .map(r -> r)
                .writeTo(localSync());
        execute().join();

        assertMap(hz(), "sthomas5@acme.com", null);

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