Adding filtering on message key

[sgaragan]

Good day,

We are pulling from a topic that has a lot of messages but we only need to look at a subset of them. Luckily, the messages have a key set that can be used to filter out the superfluous messages via a regular expression. This made me think of a useful feature for Candyfloss, the ability to define a key filter that would be applied even before the pretransform stage, to avoid even grabbing those messages and allowing for optimized message retrieval. So something like

kstream {
  input.topic.name = dev.device-json-raw-input
  discard.topic.name = dev.candyfloss-processing-discard # All messages that didn't match any step in the pipeline
  dlq.topic.name = dev.candyfloss-processing--dlq # Messages that encountered a Java exception during the processing
  state.store.name = dev.candyfloss-counters-store # Kafka state store name to save the counter values
  state.store.max.counter.cache.age = 900000 // 15 minutes
  state.store.int.counter.wrap.limit = 10000
  state.store.long.counter.wrap.limit = 10000000
  state.store.long.counter.time.ms = 300000 # allow max 5 min for counter wrap around otherwise it's a reset
  state.store.delete.scan.frequency.days = 7 # how often to trigger scanning for old unused counters and delete them
  pre.transform = pre-transform.dev.json # A pre transformation step that applies to all messages

  key.filter = "aaa.bbb.[0-9]{3}"  # <=== Regex using the Java RegEx syntax

  pipeline = {
    oc-interface-bell-normalized {
      output.topic.name = oc-interface-bell-normalized-json
      file = pipeline/oc-interface-bell-normalized.json
    }
  }
}

Thanks,
Sean