[
https://issues.apache.org/jira/browse/KAFKA-7718?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17144079#comment-17144079
]
Jorge Esteban Quilcate Otoya commented on KAFKA-7718:
-----------------------------------------------------
[~guozhang], I'd like to propose a different approach, as you mentioned this
will require changing state stores which has go through a similar path for
timestamps already.
If headers could be elevated at the `value` level, then we could operate on
them and even persist them if needed. Something like this:
{code:java}
kstream.withHeaders()
.filter((k, v) -> Arrays.equals("value".bytes(), v.headers().lastValue("key")))
.mapValues((k, v) -> {v.headers().add("foo", "bar".getBytes()); return v;})
.setHeaders((k, v) -> v.headers())
.mapValues((k, v) -> v.value())
.to("output"){code}
I drafted a KIP here,
[https://jeqo.github.io/drafts/kip-complementary-support-headers-kafka-streams/]
but would like your feedback first to check this will indeed cover what this
issue propose.
cc [~astubbs] [~mjsax]
> Allow customized header inheritance for stateful operators in DSL
> -----------------------------------------------------------------
>
> Key: KAFKA-7718
> URL: https://issues.apache.org/jira/browse/KAFKA-7718
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Guozhang Wang
> Priority: Major
> Labels: needs-kip
>
> As a follow-up work of
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-244%3A+Add+Record+Header+support+to+Kafka+Streams+Processor+API,
> we want to provide allow users to customize how record headers are inherited
> while traversing the topology at the DSL layer (at the lower-level Processor
> API layer, users are already capable for customizing and inheriting the
> headers as they forward the records to next processor nodes).
> Today the headers are implicitly inherited throughout the topology without
> any modifications within the Streams library. For stateless operators
> (filter, map, etc) this default inheritance policy should be sufficient. For
> stateful operators where multiple input records may be generating a single
> record (i.e. it is an n:1 transformations rather than 1:1 mapping), since we
> only inherit from the triggering record, which would seem to be a "random"
> choice to the users and other records' headers are lost.
> I'd propose we extend DSL to allow users to customize the headers inheritance
> policy for stateful operators, namely Joins and Aggregations. It would
> contain two parts:
> 1) On the DSL layer, I'd suggest we extend `Joined` and `Grouped` control
> object with an additional function that allows users to pass in a lambda
> function (let's say its called HeadersMerger, but name subject to discuss
> over KIP) that takes two Headers object and generated a single Headers object
> in the return value.
> 2) On the implementation layer, we need to actually store the headers at the
> materialized state store so that they can be retrieved along with the record
> for join / aggregation processor. This would be changing the state store
> value bytes organization and hence better be considered carefully. Then when
> join / aggregate processor is triggered, the Headers of both records will be
> retrieved (one from the triggering record, one read from the materialized
> state store) and then passed to the HeadersMerger. Some low-hanging
> optimizations can be considered though, e.g. if users do not have overridden
> this interface, then we can consider not reading the headers from the other
> side at all to save IO cost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
