Keyed window buffer

[Goggin]

I would like to make a keyed window buffer, I could either do this as another buffer type, or simply add into the system_window buffer.

The idea would be to have similar to a group_by_value but at buffer time. The intention of this would be to allow for waiting a time interval per key so to ensure you receive as many of the messages as possible. My use case is one of our logging systems will split messages into many parts (7-50 syslog messages per event). Each has a sequence id, a number of messages and the order in the list. The could potentially arrive in any order.

example configuration could be something like the following:

buffer:
  keyed_window:
    timestamp_mapping: root = now()
    length_mapping: root = this.count
    key_mapping: ${! meta("kafka_key") }
    size: 60s
    check: this.messages.length() ==  this.length

When checking buffer, it would run check against list of messages per key, and flush each batch. It might make most sense to have this as a new buffer type instead of just on top of the system_window buffer type.