loicgreffier commented on PR #18739:
URL: https://github.com/apache/kafka/pull/18739#issuecomment-2836610592
Tests done:
## ✅ Stateless
```java
streamsBuilder
.stream()
.peek()
.selectKey()
.mapValues() <-- Exception here
```
- Source raw key and source raw value from the input topic.
## ✅ Aggregate (Exception in Aggregator)
```java
streamsBuilder
.stream()
.peek()
.groupBy()
.aggregate(
() -> new KafkaUserAggregate(new ArrayList<>()),
new UserAggregator(), <-- Exception here
Materialized.<String, KafkaUserAggregate, KeyValueStore<Bytes,
byte[]>>as(USER_AGGREGATE_STORE)
.withKeySerde(Serdes.String())
.withValueSerde(SerdesUtils.getValueSerdes())
.withCachingEnabled());
```
- Source raw key and source raw value from the repartition topic.
- Source raw key and source raw value of the latest record that enters the
aggregator.
## ✅ Aggregate (Exception after aggregation)
```java
streamsBuilder
.stream()
.peek()
.groupByKey()
.aggregate(
() -> new KafkaUserAggregate(new ArrayList<>()),
new UserAggregator(),
Materialized.<String, KafkaUserAggregate, KeyValueStore<Bytes,
byte[]>>as(USER_AGGREGATE_STORE)
.withKeySerde(Serdes.String())
.withValueSerde(SerdesUtils.getValueSerdes())
.withCachingEnabled())
.mapValues() <-- Exception here
```
- Source raw key and source raw value from the repartition topic.
- Source raw key and source raw value of the latest record that enters the
aggregator.
- Needs the updates on `CachingKeyValueStore.java`,
`CachingSessionStore.java`, `CachingWindowStore.java`. Otherwise,
source raw key and source raw value are null.
## ✅ Left Join Stream-Stream
```java
streamsBuilder
.stream()
.selectKey()
.leftJoin(streamTwo, (key, left, right) -> { ... // Exception here });
```
- Source raw key and source raw value from the left repartition topic.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
