CREATE OR REPLACE FUNCTION ignore_unchanged_biography()
  RETURNS TRIGGER AS
$BODY$
BEGIN
  IF NEW."biography" = '__debezium_unavailable_value'
  THEN
    NEW."biography" = OLD."biography";
  END IF;

  RETURN NEW;
END;
$BODY$ LANGUAGE PLPGSQL;

CREATE TRIGGER customer_biography_trigger
BEFORE UPDATE OF "biography"
  ON customers
FOR EACH ROW
EXECUTE PROCEDURE ignore_unchanged_biography();

@ApplicationScoped
public class TopologyProducer {

    private static final Logger LOG = LoggerFactory.getLogger(TopologyProducer.class);

    static final String BIOGRAPHY_STORE = "biography-store";

    @ConfigProperty(name = "pgtoast.customers.topic")
    String customersTopic;

    @ConfigProperty(name = "pgtoast.customers.enriched.topic")
    String customersEnrichedTopic;

    @Produces
    public Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();

        StoreBuilder<KeyValueStore<JsonObject, String>> biographyStore = (1)
                Stores.keyValueStoreBuilder(
                    Stores.persistentKeyValueStore(BIOGRAPHY_STORE),
                    new JsonObjectSerde(),
                    new Serdes.StringSerde()
                );
        builder.addStateStore(biographyStore);

        builder.<JsonObject, JsonObject>stream(customersTopic) (2)
                .transformValues(ToastColumnValueProvider::new, BIOGRAPHY_STORE)
                .to(customersEnrichedTopic);

        return builder.build();
    }

    class ToastColumnValueProvider implements
            ValueTransformerWithKey<JsonObject, JsonObject, JsonObject> {

        private KeyValueStore<JsonObject, String> biographyStore;

        @Override
        @SuppressWarnings("unchecked")
        public void init(ProcessorContext context) {
            biographyStore = (KeyValueStore<JsonObject, String>) context.getStateStore(
                TopologyProducer.BIOGRAPHY_STORE);
        }

        @Override
        public JsonObject transform(JsonObject key, JsonObject value) {
            JsonObject payload = value.getJsonObject("payload");
            JsonObject newRowState = payload.getJsonObject("after");
            String biography = newRowState.getString("biography");

            if (isUnavailableValueMarker(biography)) { (3)
                String currentValue = biographyStore.get(key); (4)

                if (currentValue == null) {
                    LOG.warn("No biography value found for key '{}'", key);
                }
                else {
                    value = Json.createObjectBuilder(value) (5)
                        .add(
                            "payload",
                            Json.createObjectBuilder(payload)
                                .add(
                                    "after",
                                    Json.createObjectBuilder(newRowState).add(
                                        "biography",
                                        currentValue
                                    )
                                )
                        )
                        .build();
                }
            }
            else { (6)
                biographyStore.put(key, biography);
            }

            return value;
        }

        private boolean isUnavailableValueMarker(String value) {
            return "__debezium_unavailable_value".contentEquals(value);
        }

        @Override
        public void close() {
        }
    }
}