mjsax commented on code in PR #21237:
URL: https://github.com/apache/kafka/pull/21237#discussion_r2836841885
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java:
##########
@@ -383,7 +383,10 @@ KTable<K, V> reduce(final Reducer<V> reducer,
* Thus, {@code aggregate(Initializer, Aggregator)} can be used to compute
aggregate functions like
* count (cf. {@link #count()}).
* <p>
- * The default value serde from config will be used for serializing the
result.
+ * The key serde is inherited from the upstream processing chain (e.g.,
from a {@code Consumed} instance
+ * or a previous {@code groupBy} operation).
Review Comment:
Well, only if the key was not changed upstream. For example:
```
builder.streams(..., Consumed.with(...)).map(...).groupByKey().aggregate()
```
For this case, `map()` might change the key type and thus the key-serde from
upstream `Consumed` cannot be re-used, and we would fall back to the default
key-serde from the config.
Also, there might not be an upstream `Consumed`:
```
builder.streams(...).groupByKey().aggregate()
```
For this case, `aggregate` would also fall back to the default key-serde
from the config.
It might be best, to phrase this more generic to no get into the weeds too
much?
```
The key serde may be inherited from the upstreams processing chain, i.e.,
if a key serde was specified on an upstream operator, and the key is not
modified afterward,
the upstream key serde is recursively pushed to downstream operators.
If no key serde was specific upstream, or if a key-changing operation was
performed upstream and the key serde could not get pushed downstream, the
default key serde from the config will be used to serializer, the result.
```
--
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]
