CODEWITHSUNDEEP
apache-kafkaapache-kafka-streams
Yair Halberstadt
Yair Halberstadt

Reputation: 6861

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 Log Compacted topic.

To do so I create a stream from each topic, and merge the streams 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)

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.

Is that correct?

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

Upvotes: 0

Views: 1135

Answers (1)

Matthias J. Sax
Matthias J. Sax

Reputation: 62350

If both input topics have the same number of partitions and use the same partitioning strategy, merge() will preserve the co-partitioning.

Compare:

Upvotes: 2

Related Questions