...

table
  .suppress(untilTimeLimit(Duration.ofSeconds(30), maxRecords(1000)))
  .toStream(); // etc.

Buffer Eviction Behavior (aka Suppress Emit Behavior)

I propose to offer the following constraints:

1. at all times, the buffer enforces the #keys bound
2. at all times, the buffer enforces the size (#bytes) bound
3. at all times, the buffer enforces the time bound

Inserting a new record may violate any or all of the constraints, and advancing stream time may violate constraint 3.

If any of the constraints are violated, the buffer will evict (and emit) records until all constraints are again satisfied.

In evicting (and emitting) records, the buffer will use the following eviction strategy:

• the oldest record (by timestamp) in the buffer gets evicted

In updating a record whose key is always in the buffer, offset order is respected.

This can be visualized as the following algorithm:

process(key, value, timestamp):
  buffer.insertOrUpdate(key -> (value, timestamp))
  while(buffer.constraintsViolated()) {
    (key -> (value, timestamp)) := buffer.evictOldest()
    emit(key, value, timestamp)
  }

This spec will maintain offset ordering for all updates to a key, but may re-order events between keys.

Here are some examples to illustrate the dynamics:

Offset order is maintained when a key is updated, regardless of timestamp:

The behavior is straightforward with no late events

Enforcing a key limit of 2:

Enforcing a byte limit of 3 (each character is one byte)

Enforcing "emit after 2ms":

Note: newly added late events can be immediately evicted. 

Enforcing "emit after 2ms":

Likewise with a key constraint of 2:

And of course with a size constraint of 3 bytes as well:

Big records can push multiple events out of the buffer

Size constraint of 3 bytes:

In fact, events can be so big they don't fit in the buffer at all.

Still 3 bytes:

Rejected alternative: evicting by offset instead of timestamp.

This causes strange behavior when there is a time constraint involved:

Rejected alternative: evicting by timestamp only when the buffer is time constrained, and using offset order otherwise

Having just one eviction strategy simplifies everything: documentation, explanations, the code, the testing, etc.

I'd also argue that neither strategy is any more or less surprising for key- or size-constrained buffers. All the "flushing" behaviors above are also present in offset-ordered eviction.

Time-based eviction is perfectly legal, as we are only required to maintain a partial order over the partition (order only needs to be maintained within each key). Such partial reordering happens when we repartition anyway.

Rejected alternative (maybe a future option): "reset the timeout" when updating a key:

The current behavior is that, if you specify some timeout, say 5 minutes, any record that gets buffered is guaranteed to be emitted at the 5 minute mark, regardless of how often it's updated.  Aka, it is emitted 5 minutes after the first update.

An alternative is to effectively reset the timer on each update, so the record would be emitted 5 minutes after the last update. This behavior is analogous to session windowing; it allows suppression to turn a high-frequency "run" up updates to a record into a single emitted record.

This was rejected in the current API simply because it may suppress a record forever, if that record is updated with a frequency higher than the suppression timeout.

For the default behavior, it seemed better to go with the one that is guaranteed to emit after the specified timeout.

But this could easily be added as an alternative in the future, if there is demand for it.

Compatibility, Deprecation, and Migration Plan

...