How to solve the problem of concurrent updates of one entity from two Kafka topics (one topic overwrites changes from the other)?

The problem of concurrent updating of a single entity from two Kafka topics arises because messages from different topics can arrive in different orders, and one update overwrites another.

To solve this, the following approaches can be used:

1. Idempotency and versioning: Store the entity version or timestamp in each message. When processing an update, compare the version with the current one in the database and apply only the more recent changes.

2. Stream joining: Use Kafka Streams or another streaming framework to merge two topics into a single stream with conflict resolution logic.

3. Using aggregation functions: Instead of overwriting, aggregate changes, for example, by applying merge operations or calculating the final state.

Example in Kotlin using version comparison:

data class EntityUpdate(val id: String, val version: Long, val data: String)

fun processUpdate(current: EntityUpdate?, incoming: EntityUpdate): EntityUpdate {
    return if (current == null || incoming.version > current.version) {
        incoming
    } else {
        current
    }
}

Thus, updates with an older version will not overwrite more recent data.