@ApplicationScoped
public class TopologyProducer {

    static final String STREAM_BUFFER_NAME = "stream-buffer-state-store";
    static final String STORE_NAME = "transaction-meta-data";

    @ConfigProperty(name = "audit.context.data.topic")
    String txContextDataTopic;

    @ConfigProperty(name = "audit.vegetables.topic")
    String vegetablesTopic;

    @ConfigProperty(name = "audit.vegetables.enriched.topic")
    String vegetablesEnrichedTopic;

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

        StoreBuilder<KeyValueStore<Long, JsonObject>> streamBufferStateStore =
                Stores
                    .keyValueStoreBuilder(
                        Stores.persistentKeyValueStore(STREAM_BUFFER_NAME),
                        new Serdes.LongSerde(),
                        new JsonObjectSerde()
                    )
                    .withCachingDisabled();
            builder.addStateStore(streamBufferStateStore); (1)

        builder.globalTable(txContextDataTopic, Materialized.as(STORE_NAME)); (2)

        builder.<JsonObject, JsonObject>stream(vegetablesTopic) (3)
                .filter((id, changeEvent) -> changeEvent != null)
                .filter((id, changeEvent) -> !changeEvent.getString("op").equals("r"))
                .transform(() -> new ChangeEventEnricher(), STREAM_BUFFER_NAME)
                .to(vegetablesEnrichedTopic);

        return builder.build();
    }
}

audit.context.data.topic=dbserver1.inventory.transaction_context_data
audit.vegetables.topic=dbserver1.inventory.vegetable
audit.vegetables.enriched.topic=dbserver1.inventory.vegetable.enriched

# may be overridden with env vars
quarkus.kafka-streams.bootstrap-servers=localhost:9092
quarkus.kafka-streams.application-id=auditlog-enricher
quarkus.kafka-streams.topics=${audit.context.data.topic},${audit.vegetables.topic}

# pass-through
kafka-streams.cache.max.bytes.buffering=10240
kafka-streams.commit.interval.ms=1000
kafka-streams.metadata.max.age.ms=500
kafka-streams.auto.offset.reset=earliest
kafka-streams.metrics.recording.level=DEBUG
kafka-streams.default.key.serde=io.debezium.demos.auditing.enricher.JsonObjectSerde
kafka-streams.default.value.serde=io.debezium.demos.auditing.enricher.JsonObjectSerde
kafka-streams.processing.guarantee=exactly_once

class ChangeEventEnricher implements Transformer
        <JsonObject, JsonObject, KeyValue<JsonObject, JsonObject>> {

    private static final Long BUFFER_OFFSETS_KEY = -1L;

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

    private ProcessorContext context;
    private KeyValueStore<JsonObject, JsonObject> txMetaDataStore;
    private KeyValueStore<Long, JsonObject> streamBuffer; (5)

    @Override
    @SuppressWarnings("unchecked")
    public void init(ProcessorContext context) {
        this.context = context;
        streamBuffer = (KeyValueStore<Long, JsonObject>) context.getStateStore(
            TopologyProducer.STREAM_BUFFER_NAME
        );
        txMetaDataStore = (KeyValueStore<JsonObject, JsonObject>) context.getStateStore(
            TopologyProducer.STORE_NAME
        );

        context.schedule(
            Duration.ofSeconds(1),
            PunctuationType.WALL_CLOCK_TIME, ts -> enrichAndEmitBufferedEvents()
        ); (4)
    }

    @Override
    public KeyValue<JsonObject, JsonObject> transform(JsonObject key, JsonObject value) {
        boolean enrichedAllBufferedEvents = enrichAndEmitBufferedEvents(); (3)

        if (!enrichedAllBufferedEvents) {
            bufferChangeEvent(key, value);
            return null;
        }

        KeyValue<JsonObject, JsonObject> enriched = enrichWithTxMetaData(key, value); (1)
        if (enriched == null) { (2)
            bufferChangeEvent(key, value);
        }

        return enriched;
    }

    /**
     * Enriches the buffered change event(s) with the metadata from the associated
     * transactions and forwards them.
     *
     * @return {@code true}, if all buffered events were enriched and forwarded,
     *         {@code false} otherwise.
     */
    private boolean enrichAndEmitBufferedEvents() { (3)
        Optional<BufferOffsets> seq = bufferOffsets();

        if (!seq.isPresent()) {
            return true;
        }

        BufferOffsets sequence = seq.get();

        boolean enrichedAllBuffered = true;

        for(long i = sequence.getFirstValue(); i < sequence.getNextValue(); i++) {
            JsonObject buffered = streamBuffer.get(i);

            LOG.info("Processing buffered change event for key {}",
                    buffered.getJsonObject("key"));

            KeyValue<JsonObject, JsonObject> enriched = enrichWithTxMetaData(
                    buffered.getJsonObject("key"), buffered.getJsonObject("changeEvent"));
            if (enriched == null) {
                enrichedAllBuffered = false;
                break;
            }

            context.forward(enriched.key, enriched.value);
            streamBuffer.delete(i);
            sequence.incrementFirstValue();
        }

        if (sequence.isModified()) {
            streamBuffer.put(BUFFER_OFFSETS_KEY, sequence.toJson());
        }

        return enrichedAllBuffered;
    }

    /**
     * Adds the given change event to the stream-side buffer.
     */
    private void bufferChangeEvent(JsonObject key, JsonObject changeEvent) { (2)
        LOG.info("Buffering change event for key {}", key);

        BufferOffsets sequence = bufferOffsets().orElseGet(BufferOffsets::initial);

        JsonObject wrapper = Json.createObjectBuilder()
                .add("key", key)
                .add("changeEvent", changeEvent)
                .build();

        streamBuffer.putAll(Arrays.asList(
                KeyValue.pair(sequence.getNextValueAndIncrement(), wrapper),
                KeyValue.pair(BUFFER_OFFSETS_KEY, sequence.toJson())
        ));
    }

    /**
     * Enriches the given change event with the metadata from the associated
     * transaction.
     *
     * @return The enriched change event or {@code null} if no metadata for the
     *         associated transaction was found.
     */
    private KeyValue<JsonObject, JsonObject> enrichWithTxMetaData(JsonObject key,
            JsonObject changeEvent) { (1)
        JsonObject txId = Json.createObjectBuilder()
                .add("transaction_id", changeEvent.get("source").asJsonObject()
                        .getJsonNumber("txId").longValue())
                .build();

        JsonObject metaData = txMetaDataStore.get(txId);

        if (metaData != null) {
            LOG.info("Enriched change event for key {}", key);

            metaData = Json.createObjectBuilder(metaData.get("after").asJsonObject())
                    .remove("transaction_id")
                    .build();

            return KeyValue.pair(
                    key,
                    Json.createObjectBuilder(changeEvent)
                        .add("audit", metaData)
                        .build()
            );
        }

        LOG.warn("No metadata found for transaction {}", txId);
        return null;
    }

    private Optional<BufferOffsets> bufferOffsets() {
        JsonObject bufferOffsets = streamBuffer.get(BUFFER_OFFSETS_KEY);
        if (bufferOffsets == null) {
            return Optional.empty();
        }
        else {
            return Optional.of(BufferOffsets.fromJson(bufferOffsets));
        }
    }

    @Override
    public void close() {
    }
}

{"id":10}
{
    "before": {
        "id": 10,
        "description": "Yummy!",
        "name": "Tomato"
    },
    "after": {
        "id": 10,
        "description": "Tasty!",
        "name": "Tomato"
    },
    "source": {
        "version": "0.10.0-SNAPSHOT",
        "connector": "postgresql",
        "name": "dbserver1",
        "ts_ms": 1569700445392,
        "snapshot": "false",
        "db": "vegetablesdb",
        "schema": "inventory",
        "table": "vegetable",
        "txId": 610,
        "lsn": 34204240,
        "xmin": null
    },
    "op": "u",
    "ts_ms": 1569700445537,
    "audit": {
        "client_date": 1566461551000000,
        "usecase": "UPDATE VEGETABLE",
        "user_name": "farmermargaret"
    }
}