Does Merging two kafka-streams preserve co-partitioning

[Yair Halberstadt]

Hi
I asked this question on stack-overflow and was wondering if anyone here
could answer it:
https://stackoverflow.com/questions/59820243/does-merging-two-kafka-streams-preserve-co-partitioning


I have 2 co-partitioned kafka topics. One contains automatically generated
data, and the other manual overrides.

I want to merge them and filter out any automatically generated data that
has already been manually overidden, and then forward everything to a
combined ChangeLog topic.

To do so I create a stream from each topic, and [merge the streams](
https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KStream.html#merge-org.apache.kafka.streams.kstream.KStream-)
using the dsl API.

I then apply the following transform, which stores any manual data, and
deletes any automatic data which has already been manually overidden:
(Scala but should be pretty easy to understand if you know java)

```scala
class FilterManuallyClassifiedTransformer(manualOverridesStoreName : String)
  extends Transformer[Long, Data, KeyValue[Long, Data]] {

  // Array[Byte] used as dummy value since we don't use the value
  var store: KeyValueStore[Long, Array[Byte]] = _

  override def init(context: ProcessorContext): Unit = {
    store = context.getStateStore(manualOverridesStoreName
).asInstanceOf[KeyValueStore[Long, Array[Byte]]]
  }

  override def close(): Unit = {}

  override def transform(key: Long, value: Data): KeyValue[Long, Data] = {
    if (value.getIsManual) {
      store.put(key, Array.emptyByteArray)
      new KeyValue(key, value)
    }
    else if (store.get(key) == null) {
      new KeyValue(key, value)
    }
    else {
      null
    }
  }
}
```

If I understand correctly, there is no guarantee this will work unless
manual data and automatic data with the same key are in the same partition.
Otherwise the manual override might be stored in a different state store to
the one that the automatic data checks.

And even if they are stored in the same StateStore there might be race
conditions, where an automatic data checks the state store, then the manual
override is added to the state store, then the manual override is written
to the output topic, then the automatic data is written to the output
topic, leading to the automatic data overwriting the manual override.

Is that correct?

And if so will `merge` preserve the co-partitioning guarantee I need?

Thanks for your help