Add config to limit the total buffer size of EMIT FINAL queries

[agavra]

Is this feature related to a problem?

Queries that use EMIT FINAL (i.e. kafka streams suppression) use an in memory buffer to hold all of the key/value pairs that arrive within a window for a given source. There is a currently a limit on the size of that buffer (see #5969), but this limit is somewhat (no pun intended) limited - it puts a limit on the buffer size for each topic-partition.

This, from an operator perspective, is not enough to give any guarantees on total memory usage by suppression buffers because on startup ksqlDB cannot predict the number of topic-partitions that suppression queries will use.

If we get close to our memory utilization, the experience to the end user will be bad: either:

1. the query will die gracefully by hitting the cap and memory will be reclaimed (this is the best case scenario)
2. we hit OOM because there are too many full buffers, crashing the server
3. we hit crippling GC which will cause the queries to lag without a real course of action

Describe the solution you'd like

Option 1:

We should introduce a configuration that limits the total amount of memory that suppression buffers can use (e.g. ksql.suppress.total.buffer.quota.bytes. Each time a user issues a query, it will take up ksql.suppress.buffer.size.bytes * num_partitions away from the quota allowed for suppression buffers.

We should also rename ksql.suppress.buffer.size.bytes to indicate that this is the quota per topic-partition, not for an entire query.

Option 2:

Not exclusive option 1 above, we should have ksql.suppress.buffer.size.bytes be dynamically divided by the number of partitions when we set it in the buffer size.

Describe alternatives you've considered

• Fixing suppress to spill over onto disk. This is difficult as suppression requires range scans based on time, something RocksDB does not support
• Not having this feature means that we could size the suppression buffer to be the max heap / max queries / K where K is some constant that gives us some extra buffer. The problem is that this would size the suppress buffers to perhaps be too small to be of any real use.