[ https://issues.apache.org/jira/browse/KAFKA-15595?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Abdullah alkhawatrah updated KAFKA-15595: ----------------------------------------- Description: Hey, While upgrading to 3.5.1 from 3.2.X I noticed a change in SessionWindow aggregate behaviour, it seems now that custom headers added before the aggregate are dropped. I could reproduce the behaviour with the following test topology: {code:java} // code placeholder final StreamsBuilder builder = new StreamsBuilder(); builder.stream(inputTopic, Consumed.with(EARLIEST)) .process(() -> new Processor<Object, Object, Object, Object>() { private ProcessorContext<Object, Object> context; @Override public void init(final ProcessorContext<Object, Object> context) { this.context = context; } @Override public void process(Record<Object, Object> record) { record.headers().add("key1", record.value().toString().getBytes()); context.forward(record); } }) .groupByKey() .windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofDays(1L), Duration.ofDays(1L))) .aggregate(() -> 1, (key, value, aggregate) -> aggregate, (aggKey, aggOne, aggTwo) -> aggTwo) .toStream() .map((key, value) -> new KeyValue<>(key.key(), value)) .to(outputTopic); {code} Checking evens in the `outputTopic` show that the headers are empty. With 3.2.* the same topology would have propagated the headers. I can see here: [https://github.com/apache/kafka/blob/2c6fb6c54472e90ae17439e62540ef3cb0426fe3/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java#L205] that now a new record is created ignoring the headers, while in 3.2.2, the same record was forwarded after changing the key and value while the keeping the headers: [https://github.com/apache/kafka/blob/3.2.2/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java#L196] was: Hey, While upgrading to 3.5.1 from 3.2.X I noticed a change in SessionWindow aggregate behaviour, it seems now that custom headers added before the aggregate are dropped. I could reproduce the behaviour with the following test topology: {code:java} // code placeholder final StreamsBuilder builder = new StreamsBuilder(); builder.stream(inputTopic, Consumed.with(EARLIEST)) .process(() -> new Processor<Object, Object, Object, Object>() { private ProcessorContext<Object, Object> context; @Override public void init(final ProcessorContext<Object, Object> context) { this.context = context; } @Override public void process(Record<Object, Object> record) { record.headers().add("key1", record.value().toString().getBytes()); context.forward(record); } }) .groupByKey() .windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofDays(1L), Duration.ofDays(1L))) .aggregate(() -> 1, (key, value, aggregate) -> aggregate, (aggKey, aggOne, aggTwo) -> aggTwo) .toStream() .map((key, value) -> new KeyValue<>(key.key(), value)) .to(outputTopic); {code} Checking evens in the `outputTopic` show that the headers are empty. With 3.2.* the same topology would have propagated the headers. > Session window aggregate drops records headers > ---------------------------------------------- > > Key: KAFKA-15595 > URL: https://issues.apache.org/jira/browse/KAFKA-15595 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 3.5.1 > Reporter: Abdullah alkhawatrah > Priority: Major > > Hey, > While upgrading to 3.5.1 from 3.2.X I noticed a change in SessionWindow > aggregate behaviour, it seems now that custom headers added before the > aggregate are dropped. > I could reproduce the behaviour with the following test topology: > {code:java} > // code placeholder > final StreamsBuilder builder = new StreamsBuilder(); > builder.stream(inputTopic, Consumed.with(EARLIEST)) > .process(() -> new Processor<Object, Object, Object, Object>() { > private ProcessorContext<Object, Object> context; > @Override > public void init(final ProcessorContext<Object, Object> context) { > this.context = context; > } > @Override > public void process(Record<Object, Object> record) { > record.headers().add("key1", > record.value().toString().getBytes()); > context.forward(record); > } > }) > .groupByKey() > > .windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofDays(1L), > Duration.ofDays(1L))) > .aggregate(() -> 1, > (key, value, aggregate) -> aggregate, > (aggKey, aggOne, aggTwo) -> aggTwo) > .toStream() > .map((key, value) -> new KeyValue<>(key.key(), value)) > .to(outputTopic); {code} > Checking evens in the `outputTopic` show that the headers are empty. With > 3.2.* the same topology would have propagated the headers. > > I can see here: > [https://github.com/apache/kafka/blob/2c6fb6c54472e90ae17439e62540ef3cb0426fe3/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java#L205] > that now a new record is created ignoring the headers, while in 3.2.2, the > same record was forwarded after changing the key and value while the keeping > the headers: > [https://github.com/apache/kafka/blob/3.2.2/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java#L196] > -- This message was sent by Atlassian Jira (v8.20.10#820010)