mjsax commented on code in PR #23007: URL: https://github.com/apache/kafka/pull/23007#discussion_r3693063199
########## docs/streams/developer-guide/config-streams.md: ########## @@ -1407,7 +1407,7 @@ Serde for the inner class of a windowed record. Must implement the `Serde` inter > > See [KIP-1271](https://cwiki.apache.org/confluence/x/QIM8G) for migration > procedures, changelog compatibility, restore behavior, and per-record > overhead. > -> **Current limitations**: `dsl.store.format=HEADERS` changes the state store format. It does not define how DSL operators create headers for output records. Some operators write empty headers to their materialized stores, and the buffer stores used by `suppress()` and left/outer stream-stream joins are not headers-aware. See [Stateful transformations](/{version}/streams/developer-guide/dsl-api.html#stateful-transformations) and the [Streams upgrade guide](/{version}/streams/upgrade-guide.html#current-limitations) for details. +> **Current limitations**: `dsl.store.format=HEADERS` changes the state store format. It does not define how DSL operators create headers for output records. Some operators write empty headers to their materialized stores, and the buffer store used by left/outer stream-stream joins for not-yet-matched records is not headers-aware. The `suppress()` buffer is headers-aware under `HEADERS`: it preserves the headers of the records it buffers. See [Stateful transformations](/{version}/streams/developer-guide/dsl-api.html#stateful-transformations) and the [Streams upgrade guide](/{version}/streams/upgrade-guide.html#current-limitations) for details. Review Comment: ```suggestion > **Current limitations**: `dsl.store.format=HEADERS` changes the state store format. It does not define how DSL operators create headers for output records. Some operators write empty headers to their materialized stores, and the buffer store used by left/outer stream-stream joins for not-yet-matched records is not headers-aware. See [Stateful transformations](/{version}/streams/developer-guide/dsl-api.html#stateful-transformations) and the [Streams upgrade guide](/{version}/streams/upgrade-guide.html#current-limitations) for details. ``` ########## docs/streams/upgrade-guide.md: ########## @@ -131,14 +131,17 @@ Today, DSL result headers behave as follows: * Aggregations (`count`, `reduce`, `aggregate`, including their windowed and session-windowed variants), KTable-KTable joins (inner / left / outer), materialized `KTable.mapValues`, `KStream.toTable()`, and `StreamsBuilder.table()` write empty headers to their materialized stores. * KStream-KStream join window stores keep source-record headers, but join result records do not get computed or merged headers. They may carry the headers from the record that triggered the result. -* `suppress()` and left/outer stream-stream joins use non-headers-aware buffer stores. Records that pass through those buffers lose their headers. +* `suppress()` uses a headers-aware buffer store under `dsl.store.format=HEADERS`. Each buffered value is stored together with the headers of the record it came from, and the record emitted when a buffered row is evicted carries the headers of the value being emitted. Under `DEFAULT`, the buffer stores plain values plus a single record context per buffered row, so a row updated by a later record before eviction preserves only the headers of that later record. +* Left/outer stream-stream joins use a non-headers-aware buffer store for not-yet-matched records. Records that pass through that buffer lose their headers. A follow-up KIP will give users explicit control over how DSL result headers are computed. See [Stateful transformations](/{version}/streams/developer-guide/dsl-api.html#stateful-transformations) for more details. #### Changelog, migration, and performance KIP-1285 does not change the changelog wire format, the migration procedure, or the per-store overhead — those are properties of the underlying KIP-1271 stores. See the [KIP-1271 section](#kip-1271-headers-aware-stores) above for the full description of changelog compatibility, the lazy per-key RocksDB migration on `DEFAULT`→`HEADERS`, the restore behavior, and the per-record size impact. The DSL config `dsl.store.format` only controls which operators participate; once an operator is using a headers-aware store, the store runtime behavior is identical to the Processor API case. +The `suppress()` buffer is the one exception worth noting: under `dsl.store.format=HEADERS` its changelog records keep the existing V3 value format and carry the headers of the buffered old and prior values in additional Kafka record headers (`vh.old`, `vh.prior`). Older versions ignore those headers and restore the values without them, so the changelog stays readable across a downgrade. Review Comment: Same. 4.3 section should not get updated, because nothing changed for 4.3. ########## docs/streams/upgrade-guide.md: ########## @@ -131,14 +131,17 @@ Today, DSL result headers behave as follows: * Aggregations (`count`, `reduce`, `aggregate`, including their windowed and session-windowed variants), KTable-KTable joins (inner / left / outer), materialized `KTable.mapValues`, `KStream.toTable()`, and `StreamsBuilder.table()` write empty headers to their materialized stores. * KStream-KStream join window stores keep source-record headers, but join result records do not get computed or merged headers. They may carry the headers from the record that triggered the result. -* `suppress()` and left/outer stream-stream joins use non-headers-aware buffer stores. Records that pass through those buffers lose their headers. +* `suppress()` uses a headers-aware buffer store under `dsl.store.format=HEADERS`. Each buffered value is stored together with the headers of the record it came from, and the record emitted when a buffered row is evicted carries the headers of the value being emitted. Under `DEFAULT`, the buffer stores plain values plus a single record context per buffered row, so a row updated by a later record before eviction preserves only the headers of that later record. +* Left/outer stream-stream joins use a non-headers-aware buffer store for not-yet-matched records. Records that pass through that buffer lose their headers. Review Comment: This is the 4.3 section, and for 4.3 nothing changed. So this should not be modified IMHO. -- We should rather add a bullet point to the 4.4 section, calling out that `suppress()` does now support headers via KIP-1285) ########## docs/streams/developer-guide/dsl-api.md: ########## @@ -4673,7 +4674,7 @@ The key parts of this program are: One thing to note is that suppression is just like any other Kafka Streams operator, so you can build a topology with two branches emerging from the `count`, one suppressed, and one not, or even multiple differently configured suppressions. This allows you to apply suppressions where they are needed and otherwise rely on the default continuous update behavior. -**Note on headers-aware state stores:** `suppress()` uses an in-memory buffer that is not headers-aware. Record headers attached to upstream records are not preserved across the suppression boundary, even when [`dsl.store.format=HEADERS`](../config-streams#dsl-store-format) is set globally per [KIP-1285](https://cwiki.apache.org/confluence/x/4ow8G). +**Note on headers-aware state stores:** With [`dsl.store.format=HEADERS`](../config-streams#dsl-store-format) set per [KIP-1285](https://cwiki.apache.org/confluence/x/4ow8G), the in-memory buffer used by `suppress()` is headers-aware: each buffered value is stored together with the headers of the record it came from, so record headers are preserved across the suppression boundary, and the record emitted when a buffered row is evicted carries the headers of the value being emitted. With the default `dsl.store.format=DEFAULT`, the buffer stores plain values plus a single record context per buffered row, so a row that is updated by a later record before it is evicted preserves only the headers of that later record. Review Comment: Similar to above. Should we just drop the entire paragraph? ########## docs/streams/developer-guide/config-streams.md: ########## @@ -1407,7 +1407,7 @@ Serde for the inner class of a windowed record. Must implement the `Serde` inter > > See [KIP-1271](https://cwiki.apache.org/confluence/x/QIM8G) for migration > procedures, changelog compatibility, restore behavior, and per-record > overhead. > -> **Current limitations**: `dsl.store.format=HEADERS` changes the state store format. It does not define how DSL operators create headers for output records. Some operators write empty headers to their materialized stores, and the buffer stores used by `suppress()` and left/outer stream-stream joins are not headers-aware. See [Stateful transformations](/{version}/streams/developer-guide/dsl-api.html#stateful-transformations) and the [Streams upgrade guide](/{version}/streams/upgrade-guide.html#current-limitations) for details. +> **Current limitations**: `dsl.store.format=HEADERS` changes the state store format. It does not define how DSL operators create headers for output records. Some operators write empty headers to their materialized stores, and the buffer store used by left/outer stream-stream joins for not-yet-matched records is not headers-aware. The `suppress()` buffer is headers-aware under `HEADERS`: it preserves the headers of the records it buffers. See [Stateful transformations](/{version}/streams/developer-guide/dsl-api.html#stateful-transformations) and the [Streams upgrade guide](/{version}/streams/upgrade-guide.html#current-limitations) for details. Review Comment: The bullet point is "Current limitations" so might be better to not mention suppress() at all? There is no limitation for it any longer. The only thing we could call out would be https://issues.apache.org/jira/browse/KAFKA-20856 but not sure if it's worth it? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
