You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Jorge Esteban Quilcate Otoya <qu...@gmail.com> on 2022/03/21 20:13:03 UTC
Re: [DISCUSS] KIP-634: Complementary support for headers in Kafka Streams DSL
Hi all,
With the acceptance of KIP-820 which will enable easier access to the
Record's metadata and headers and the potential design of a new version of
the DSL, I will set this KIP as inactive/dormant for the time being.
Thanks, everyone for the great discussions!
Jorge.
On Wed, 16 Feb 2022 at 00:08, Matthias J. Sax <mj...@apache.org> wrote:
> Sorry for playing devil's advocate, but do we really think it's a good
> way to design the API? To me, if feels like a cumbersome workaround.
>
> Personally, I believe that we are hitting a point for the DSL that
> requires a redesign from scratch. When the DSL was designed 5 years ago,
> record timestamp was just newly added (and did not play a significant
> role yet), and there was no record headers. That's why we have a
> kv-based model with `KStream<K,V>` and `KTable<K,V>` types.
>
> Given the changes in Kafka (and Kafka Streams) that accumulated over the
> last 5 years, it seems that a better API would be to have a
> KStream<Record<K,V>>` and `KTable<Record<K,V>>` model.
>
> I also think that we should clearly separate (modifiable) data (key,
> value, timestamp, headers) from (immutable) meta-data (topic name,
> partition, offset). And to go one step further, I am not even sure it
> meta-data makes much sense for non-source records (e.g., `offset` is
> more or less useless after a flatMap(), aggregation() -- similar for
> topic and partition after a join()). It would make sense to me, to take
> considerations like this into account designing the API.
>
> I would propose to not move forward with a "hacky design" (sorry for the
> strong terminology...) but to starting a DSL 2.0 discussion to come up
> with a long term sensible re-design. Of course, it would be a much
> larger effort, and might not provide a short term fix.
>
> A DSL 2.0 would not only allow us to add support for headers, but also
> to fix many other issue in the DSL that are not easily fixed without
> breaking compatibility (one example is KIP-300:
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-300%3A+Add+Windowed+KTable+API+in+StreamsBuilder
> )
>
> Thoughts?
>
>
> -Matthias
>
> On 2/15/22 11:55 AM, John Roesler wrote:
> > Thanks for the update, Jorge,
> >
> > I've just looked over the KIP again. Just one more small
> > concern:
> >
> > 5) We can't just change the type of Record#headers() to a
> > new fully qualified type. That would be a source-
> > incompatible breaking change for users.
> >
> > Out options are:
> > * Deprecate the existing method and create a new one with
> > the new type
> > * If the existing Headers is "not great but ok", then maybe
> > we leave it alone.
> >
> > Thanks,
> > -John
> >
> >
> > On Fri, 2022-02-11 at 20:40 +0000, Jorge Esteban Quilcate
> > Otoya wrote:
> >> John and team,
> >>
> >> The following changes have been applied to the KIP following your
> feedback:
> >>
> >> - Leverage `Record<K, V>` instead of introducing a new type
> >> (`RecordValue<V>`).
> >> - `RecordSerde<K, V>` for stateful operations using `Record<K, V>` as
> value.
> >> - Extend `Record<K, V>` to:
> >> - Implement `RecordMetadata` to expose `topic`, `partition`, and
> `offset`
> >> - Use `Headers` abstraction introduce on this KIP instead of core one
> >>
> >> KIP:
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+and+record+metadata+in+Kafka+Streams+DSL
> >>
> >> Looking forward to your feedback.
> >>
> >> Have a great weekend!
> >>
> >> On Thu, 10 Feb 2022 at 13:15, Jorge Esteban Quilcate Otoya <
> >> quilcate.jorge@gmail.com> wrote:
> >>
> >>>> What do you think about instead adding topic and
> >>> partition to Record?
> >>>
> >>> This is a very interesting idea. Forgot to consider this addition from
> >>> KIP-478.
> >>>
> >>> `Record` would also require `offset`. Maybe implementing
> `RecordMetadata`
> >>> and adding these fields as part of the constructor to keep them
> immutable
> >>> in comparison to the other fields?
> >>> It would also need to change `Record`'s headers type to the new one
> >>> proposed on this KIP.
> >>>
> >>> Let me explore this approach in more detail and update the KIP.
> >>>
> >>>> I find the name "mapRecordValue" to be a bit confusing
> >>> because it seems like it would map the value of a record.
> >>> What do you think about "mapValueToRecord" instead?
> >>>
> >>> Agree. It will depend on how we solve 1). If we end up using `Record`
> then
> >>> `mapValueToRecord` will make even more sense.
> >>>
> >>>> I agree with adding the serde explicitly. However, it
> >>> would be good to state whether and when we'll automatically
> >>> wrap a value serde. For example, if the value serde is known
> >>> (or if we're using a default serde from the config), will
> >>> Streams automatically wrap it downstream of the record-
> >>> mapping operator?
> >>>
> >>> Good point. The goal is as you describe it: only when
> `mapValueToRecord`
> >>> is called, the Serde will be implicitly applied.
> >>> Will make this explicit on the KIP.
> >>>
> >>>
> >>> On Wed, 9 Feb 2022 at 20:05, John Roesler <vv...@apache.org> wrote:
> >>>
> >>>> Hello Jorge,
> >>>>
> >>>> Thanks for bringing this up again!
> >>>>
> >>>> I've just read over the current version of the KIP.
> >>>>
> >>>> 1) I wonder if we really need RecordValue, since we now have
> >>>> Record, and they are almost the same, both in API and in
> >>>> purpose. What do you think about instead adding topic and
> >>>> partition to Record?
> >>>>
> >>>> 2) I find the name "mapRecordValue" to be a bit confusing
> >>>> because it seems like it would map the value of a record.
> >>>> What do you think about "mapValueToRecord" instead?
> >>>>
> >>>> 3) I agree with adding the serde explicitly. However, it
> >>>> would be good to state whether and when we'll automatically
> >>>> wrap a value serde. For example, if the value serde is known
> >>>> (or if we're using a default serde from the config), will
> >>>> Streams automatically wrap it downstream of the record-
> >>>> mapping operator?
> >>>>
> >>>> Otherwise, your proposal looks good to me!
> >>>>
> >>>> Thanks,
> >>>> -John
> >>>>
> >>>> On Tue, 2022-02-08 at 18:06 +0000, Jorge Esteban Quilcate
> >>>> Otoya wrote:
> >>>>> Hi Dev team,
> >>>>>
> >>>>> I'd like to revamp the KIP again:
> >>>>>
> >>>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+and+record+metadata+in+Kafka+Streams+DSL
> >>>>>
> >>>>> - Reference implementation is now using the latest `Processor` API
> from
> >>>>> KIP-478: https://github.com/apache/kafka/pull/10265/files for both
> >>>>> Processors backing changes on the KStream API.
> >>>>> - It is proposing to still extend `To` class for backwards
> >>>> compatibility.
> >>>>>
> >>>>> Looking forward to your feedback.
> >>>>>
> >>>>> Regards,
> >>>>> Jorge.
> >>>>>
> >>>>> On Thu, 4 Mar 2021 at 18:38, Jorge Esteban Quilcate Otoya <
> >>>>> quilcate.jorge@gmail.com> wrote:
> >>>>>
> >>>>>> Hi everyone!
> >>>>>>
> >>>>>> I'd like to revamp this KIP. I have made some significant changes on
> >>>> the
> >>>>>> scope:
> >>>>>> - Added `mapRecordValue` to map not only headers, but other record
> >>>>>> metadata: topic name, partition, offset, and timestamp into a new
> type
> >>>>>> `RecordValue<V>`.
> >>>>>> - Added a serde for `RecordValue` to support stateful operations.
> >>>>>> - Added `setRecordHeaders` to apply headers to record crossing the
> >>>> stream.
> >>>>>> - Added headers to `To` to update headers via `context.forward(k, v,
> >>>> to)`.
> >>>>>>
> >>>>>> New link:
> >>>>>>
> >>>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+and+record+metadata+in+Kafka+Streams+DSL
> >>>>>>
> >>>>>> Looking forward to your feedback,
> >>>>>>
> >>>>>> Cheers and stay safe,
> >>>>>> Jorge.
> >>>>>>
> >>>>>> On Thu, Oct 29, 2020 at 12:33 AM Jorge Esteban Quilcate Otoya <
> >>>>>> quilcate.jorge@gmail.com> wrote:
> >>>>>>
> >>>>>>> Thanks Sophie! Haven't followed KIP-478 but sounds great.
> >>>>>>> I'll be happy to help on that migration to the new PAPI if it's
> >>>> still an
> >>>>>>> open issue. We can bump this KIP after that.
> >>>>>>>
> >>>>>>> Cheers,
> >>>>>>> Jorge.
> >>>>>>>
> >>>>>>> On Wed, Oct 28, 2020 at 7:00 PM Sophie Blee-Goldman <
> >>>> sophie@confluent.io>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>> I *think* that the `To` Matthias was referring to was not
> >>>> KStream#to but
> >>>>>>>> the To class
> >>>>>>>> which is accepted as a possible parameter of
> >>>> ProcessorContext#forward
> >>>>>>>> (correct
> >>>>>>>> me if wrong).
> >>>>>>>>
> >>>>>>>> This was on the old ProcessorContext interface, which has now been
> >>>>>>>> replaced with
> >>>>>>>> the new api.ProcessorContext in KIP-478. In the new interface
> >>>> we've moved
> >>>>>>>> away
> >>>>>>>> from the forward signatures that accept a separate key or value or
> >>>>>>>> timestamp or To,
> >>>>>>>> and wrapped all of these into a single Record class. This new
> >>>> Record
> >>>>>>>> class
> >>>>>>>> has the
> >>>>>>>> headers as a field, so it seems like KIP-478 has happened to
> >>>> solve the
> >>>>>>>> lack
> >>>>>>>> of support
> >>>>>>>> for Headers in the PAPI along the way.
> >>>>>>>>
> >>>>>>>> This is all somewhat recent, and probably wasn't yet sorted out
> >>>> at the
> >>>>>>>> time
> >>>>>>>> of Matthias'
> >>>>>>>> last reply. But given how this worked out it seems like we can
> >>>> just focus
> >>>>>>>> on adding
> >>>>>>>> support for Headers in the DSL in this KIP by building off of the
> >>>>>>>> groundwork of
> >>>>>>>> KIP-478? It doesn't seem necessary to go back and add support for
> >>>> headers
> >>>>>>>> in the old
> >>>>>>>> PAPI, since this will (or already has?) been deprecated.
> >>>>>>>>
> >>>>>>>> The one challenge is that this will presumably require that we
> >>>> migrate
> >>>>>>>> all
> >>>>>>>> DSL operators
> >>>>>>>> to the new PAPI before adding header support for those operators.
> >>>> But
> >>>>>>>> that
> >>>>>>>> definitely
> >>>>>>>> sounds achievable here
> >>>>>>>>
> >>>>>>>> On Wed, Oct 28, 2020 at 11:10 AM Jorge Esteban Quilcate Otoya <
> >>>>>>>> quilcate.jorge@gmail.com> wrote:
> >>>>>>>>
> >>>>>>>>> Hi Matthias,
> >>>>>>>>>
> >>>>>>>>> Sorry for the late reply.
> >>>>>>>>>
> >>>>>>>>> I like the proposal. Just to check if I got it right:
> >>>>>>>>>
> >>>>>>>>> We can extend the `kstream.to()` function to support setting
> >>>> headers.
> >>>>>>>>> e.g.:
> >>>>>>>>>
> >>>>>>>>> ```
> >>>>>>>>> void to(final String topic,
> >>>>>>>>> final Produced<K, V> produced,
> >>>>>>>>> final HeadersExtractor<K, V> headersExtractor);
> >>>>>>>>> ```
> >>>>>>>>>
> >>>>>>>>> where `HeadersExtractor`:
> >>>>>>>>>
> >>>>>>>>> ```
> >>>>>>>>> public interface HeadersExtractor<K, V> {
> >>>>>>>>> Headers extract(final K key, final V value, final
> >>>> RecordContext
> >>>>>>>>> recordContext);
> >>>>>>>>> }
> >>>>>>>>> ```
> >>>>>>>>>
> >>>>>>>>> This would require to change `Topology#addSink()` to support
> >>>> this
> >>>>>>>>> extractor as well.
> >>>>>>>>>
> >>>>>>>>> If this is aligned with your proposal, I'm happy to add it to
> >>>> this KIP.
> >>>>>>>>>
> >>>>>>>>> Cheers,
> >>>>>>>>> Jorge.
> >>>>>>>>>
> >>>>>>>>> On Fri, Sep 11, 2020 at 11:03 PM Matthias J. Sax <
> >>>> mjsax@apache.org>
> >>>>>>>> wrote:
> >>>>>>>>>
> >>>>>>>>>> Jorge,
> >>>>>>>>>>
> >>>>>>>>>> thanks a lot for this KIP. Being able to modify headers is a
> >>>> very
> >>>>>>>>>> valuable feature.
> >>>>>>>>>>
> >>>>>>>>>> However, before we actually expose them in the DSL, I am
> >>>> wondering
> >>>>>>>> if we
> >>>>>>>>>> should improve how headers can be modified in the PAPI?
> >>>> Currently,
> >>>>>>>> it is
> >>>>>>>>>> possible but very clumsy to work with headers in the
> >>>> Processor API,
> >>>>>>>>>> because of two reasons:
> >>>>>>>>>>
> >>>>>>>>>> (1) There is no default `Headers` implementation in the
> >>>> public API
> >>>>>>>>>> (2) There is no explicit way to set headers for output
> >>>> records
> >>>>>>>>>>
> >>>>>>>>>> Currently, the input record headers are copied into the output
> >>>>>>>> records
> >>>>>>>>>> when `forward()` is called, however, it's not really a deep
> >>>> copy but
> >>>>>>>> we
> >>>>>>>>>> just copy the reference. This implies that one needs to work
> >>>> with a
> >>>>>>>>>> single mutable object that flows through multiple processors
> >>>> making
> >>>>>>>> it
> >>>>>>>>>> very error prone.
> >>>>>>>>>>
> >>>>>>>>>> Furthermore, if you want to emit multiple output records, and
> >>>> for
> >>>>>>>>>> example want to add two different headers to the output record
> >>>>>>>> (based on
> >>>>>>>>>> the same input headers), you would need to do something like
> >>>> this:
> >>>>>>>>>>
> >>>>>>>>>> Headers h = context.headers();
> >>>>>>>>>> h.add(...);
> >>>>>>>>>> context.forward(...);
> >>>>>>>>>> // remove the header you added for the first output record
> >>>>>>>>>> h.remove(...);
> >>>>>>>>>> h.add(...);
> >>>>>>>>>> context.forward(...);
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> Maybe we could extend `To` to allow passing in a new
> >>>> `Headers` object
> >>>>>>>>>> (or an `Iterable<Header>` similar to `ProducerRecord`)? We
> >>>> could
> >>>>>>>> either
> >>>>>>>>>> add it to your KIP or do a new KIP just for the PAPI.
> >>>>>>>>>>
> >>>>>>>>>> Thoughts?
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> -Matthias
> >>>>>>>>>>
> >>>>>>>>>> On 7/16/20 4:05 PM, Jorge Esteban Quilcate Otoya wrote:
> >>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>
> >>>>>>>>>>> Bumping this thread to check if there's any feedback.
> >>>>>>>>>>>
> >>>>>>>>>>> Cheers,
> >>>>>>>>>>> Jorge.
> >>>>>>>>>>>
> >>>>>>>>>>> On Tue, Jun 30, 2020 at 12:46 AM Jorge Esteban Quilcate
> >>>> Otoya <
> >>>>>>>>>>> quilcate.jorge@gmail.com> wrote:
> >>>>>>>>>>>
> >>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>
> >>>>>>>>>>>> I would like to start the discussion for KIP-634:
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+in+Kafka+Streams+DSL
> >>>>>>>>>>>>
> >>>>>>>>>>>> Looking forward to your feedback.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Thanks!
> >>>>>>>>>>>> Jorge.
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>
> >>>>
> >
>
Re: [DISCUSS] KIP-634: Complementary support for headers in Kafka Streams DSL
Posted by "Matthias J. Sax" <mj...@apache.org>.
Moving this KIP into "discarded" state. While header support is still
limited, we should rather move a `Record` base API following the new
PAPI, and thus this KIP does not make any sense any longer from my POV.
On 3/21/22 1:13 PM, Jorge Esteban Quilcate Otoya wrote:
> Hi all,
>
> With the acceptance of KIP-820 which will enable easier access to the
> Record's metadata and headers and the potential design of a new version of
> the DSL, I will set this KIP as inactive/dormant for the time being.
>
> Thanks, everyone for the great discussions!
> Jorge.
>
> On Wed, 16 Feb 2022 at 00:08, Matthias J. Sax <mj...@apache.org> wrote:
>
>> Sorry for playing devil's advocate, but do we really think it's a good
>> way to design the API? To me, if feels like a cumbersome workaround.
>>
>> Personally, I believe that we are hitting a point for the DSL that
>> requires a redesign from scratch. When the DSL was designed 5 years ago,
>> record timestamp was just newly added (and did not play a significant
>> role yet), and there was no record headers. That's why we have a
>> kv-based model with `KStream<K,V>` and `KTable<K,V>` types.
>>
>> Given the changes in Kafka (and Kafka Streams) that accumulated over the
>> last 5 years, it seems that a better API would be to have a
>> KStream<Record<K,V>>` and `KTable<Record<K,V>>` model.
>>
>> I also think that we should clearly separate (modifiable) data (key,
>> value, timestamp, headers) from (immutable) meta-data (topic name,
>> partition, offset). And to go one step further, I am not even sure it
>> meta-data makes much sense for non-source records (e.g., `offset` is
>> more or less useless after a flatMap(), aggregation() -- similar for
>> topic and partition after a join()). It would make sense to me, to take
>> considerations like this into account designing the API.
>>
>> I would propose to not move forward with a "hacky design" (sorry for the
>> strong terminology...) but to starting a DSL 2.0 discussion to come up
>> with a long term sensible re-design. Of course, it would be a much
>> larger effort, and might not provide a short term fix.
>>
>> A DSL 2.0 would not only allow us to add support for headers, but also
>> to fix many other issue in the DSL that are not easily fixed without
>> breaking compatibility (one example is KIP-300:
>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-300%3A+Add+Windowed+KTable+API+in+StreamsBuilder
>> )
>>
>> Thoughts?
>>
>>
>> -Matthias
>>
>> On 2/15/22 11:55 AM, John Roesler wrote:
>>> Thanks for the update, Jorge,
>>>
>>> I've just looked over the KIP again. Just one more small
>>> concern:
>>>
>>> 5) We can't just change the type of Record#headers() to a
>>> new fully qualified type. That would be a source-
>>> incompatible breaking change for users.
>>>
>>> Out options are:
>>> * Deprecate the existing method and create a new one with
>>> the new type
>>> * If the existing Headers is "not great but ok", then maybe
>>> we leave it alone.
>>>
>>> Thanks,
>>> -John
>>>
>>>
>>> On Fri, 2022-02-11 at 20:40 +0000, Jorge Esteban Quilcate
>>> Otoya wrote:
>>>> John and team,
>>>>
>>>> The following changes have been applied to the KIP following your
>> feedback:
>>>>
>>>> - Leverage `Record<K, V>` instead of introducing a new type
>>>> (`RecordValue<V>`).
>>>> - `RecordSerde<K, V>` for stateful operations using `Record<K, V>` as
>> value.
>>>> - Extend `Record<K, V>` to:
>>>> - Implement `RecordMetadata` to expose `topic`, `partition`, and
>> `offset`
>>>> - Use `Headers` abstraction introduce on this KIP instead of core one
>>>>
>>>> KIP:
>>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+and+record+metadata+in+Kafka+Streams+DSL
>>>>
>>>> Looking forward to your feedback.
>>>>
>>>> Have a great weekend!
>>>>
>>>> On Thu, 10 Feb 2022 at 13:15, Jorge Esteban Quilcate Otoya <
>>>> quilcate.jorge@gmail.com> wrote:
>>>>
>>>>>> What do you think about instead adding topic and
>>>>> partition to Record?
>>>>>
>>>>> This is a very interesting idea. Forgot to consider this addition from
>>>>> KIP-478.
>>>>>
>>>>> `Record` would also require `offset`. Maybe implementing
>> `RecordMetadata`
>>>>> and adding these fields as part of the constructor to keep them
>> immutable
>>>>> in comparison to the other fields?
>>>>> It would also need to change `Record`'s headers type to the new one
>>>>> proposed on this KIP.
>>>>>
>>>>> Let me explore this approach in more detail and update the KIP.
>>>>>
>>>>>> I find the name "mapRecordValue" to be a bit confusing
>>>>> because it seems like it would map the value of a record.
>>>>> What do you think about "mapValueToRecord" instead?
>>>>>
>>>>> Agree. It will depend on how we solve 1). If we end up using `Record`
>> then
>>>>> `mapValueToRecord` will make even more sense.
>>>>>
>>>>>> I agree with adding the serde explicitly. However, it
>>>>> would be good to state whether and when we'll automatically
>>>>> wrap a value serde. For example, if the value serde is known
>>>>> (or if we're using a default serde from the config), will
>>>>> Streams automatically wrap it downstream of the record-
>>>>> mapping operator?
>>>>>
>>>>> Good point. The goal is as you describe it: only when
>> `mapValueToRecord`
>>>>> is called, the Serde will be implicitly applied.
>>>>> Will make this explicit on the KIP.
>>>>>
>>>>>
>>>>> On Wed, 9 Feb 2022 at 20:05, John Roesler <vv...@apache.org> wrote:
>>>>>
>>>>>> Hello Jorge,
>>>>>>
>>>>>> Thanks for bringing this up again!
>>>>>>
>>>>>> I've just read over the current version of the KIP.
>>>>>>
>>>>>> 1) I wonder if we really need RecordValue, since we now have
>>>>>> Record, and they are almost the same, both in API and in
>>>>>> purpose. What do you think about instead adding topic and
>>>>>> partition to Record?
>>>>>>
>>>>>> 2) I find the name "mapRecordValue" to be a bit confusing
>>>>>> because it seems like it would map the value of a record.
>>>>>> What do you think about "mapValueToRecord" instead?
>>>>>>
>>>>>> 3) I agree with adding the serde explicitly. However, it
>>>>>> would be good to state whether and when we'll automatically
>>>>>> wrap a value serde. For example, if the value serde is known
>>>>>> (or if we're using a default serde from the config), will
>>>>>> Streams automatically wrap it downstream of the record-
>>>>>> mapping operator?
>>>>>>
>>>>>> Otherwise, your proposal looks good to me!
>>>>>>
>>>>>> Thanks,
>>>>>> -John
>>>>>>
>>>>>> On Tue, 2022-02-08 at 18:06 +0000, Jorge Esteban Quilcate
>>>>>> Otoya wrote:
>>>>>>> Hi Dev team,
>>>>>>>
>>>>>>> I'd like to revamp the KIP again:
>>>>>>>
>>>>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+and+record+metadata+in+Kafka+Streams+DSL
>>>>>>>
>>>>>>> - Reference implementation is now using the latest `Processor` API
>> from
>>>>>>> KIP-478: https://github.com/apache/kafka/pull/10265/files for both
>>>>>>> Processors backing changes on the KStream API.
>>>>>>> - It is proposing to still extend `To` class for backwards
>>>>>> compatibility.
>>>>>>>
>>>>>>> Looking forward to your feedback.
>>>>>>>
>>>>>>> Regards,
>>>>>>> Jorge.
>>>>>>>
>>>>>>> On Thu, 4 Mar 2021 at 18:38, Jorge Esteban Quilcate Otoya <
>>>>>>> quilcate.jorge@gmail.com> wrote:
>>>>>>>
>>>>>>>> Hi everyone!
>>>>>>>>
>>>>>>>> I'd like to revamp this KIP. I have made some significant changes on
>>>>>> the
>>>>>>>> scope:
>>>>>>>> - Added `mapRecordValue` to map not only headers, but other record
>>>>>>>> metadata: topic name, partition, offset, and timestamp into a new
>> type
>>>>>>>> `RecordValue<V>`.
>>>>>>>> - Added a serde for `RecordValue` to support stateful operations.
>>>>>>>> - Added `setRecordHeaders` to apply headers to record crossing the
>>>>>> stream.
>>>>>>>> - Added headers to `To` to update headers via `context.forward(k, v,
>>>>>> to)`.
>>>>>>>>
>>>>>>>> New link:
>>>>>>>>
>>>>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+and+record+metadata+in+Kafka+Streams+DSL
>>>>>>>>
>>>>>>>> Looking forward to your feedback,
>>>>>>>>
>>>>>>>> Cheers and stay safe,
>>>>>>>> Jorge.
>>>>>>>>
>>>>>>>> On Thu, Oct 29, 2020 at 12:33 AM Jorge Esteban Quilcate Otoya <
>>>>>>>> quilcate.jorge@gmail.com> wrote:
>>>>>>>>
>>>>>>>>> Thanks Sophie! Haven't followed KIP-478 but sounds great.
>>>>>>>>> I'll be happy to help on that migration to the new PAPI if it's
>>>>>> still an
>>>>>>>>> open issue. We can bump this KIP after that.
>>>>>>>>>
>>>>>>>>> Cheers,
>>>>>>>>> Jorge.
>>>>>>>>>
>>>>>>>>> On Wed, Oct 28, 2020 at 7:00 PM Sophie Blee-Goldman <
>>>>>> sophie@confluent.io>
>>>>>>>>> wrote:
>>>>>>>>>
>>>>>>>>>> I *think* that the `To` Matthias was referring to was not
>>>>>> KStream#to but
>>>>>>>>>> the To class
>>>>>>>>>> which is accepted as a possible parameter of
>>>>>> ProcessorContext#forward
>>>>>>>>>> (correct
>>>>>>>>>> me if wrong).
>>>>>>>>>>
>>>>>>>>>> This was on the old ProcessorContext interface, which has now been
>>>>>>>>>> replaced with
>>>>>>>>>> the new api.ProcessorContext in KIP-478. In the new interface
>>>>>> we've moved
>>>>>>>>>> away
>>>>>>>>>> from the forward signatures that accept a separate key or value or
>>>>>>>>>> timestamp or To,
>>>>>>>>>> and wrapped all of these into a single Record class. This new
>>>>>> Record
>>>>>>>>>> class
>>>>>>>>>> has the
>>>>>>>>>> headers as a field, so it seems like KIP-478 has happened to
>>>>>> solve the
>>>>>>>>>> lack
>>>>>>>>>> of support
>>>>>>>>>> for Headers in the PAPI along the way.
>>>>>>>>>>
>>>>>>>>>> This is all somewhat recent, and probably wasn't yet sorted out
>>>>>> at the
>>>>>>>>>> time
>>>>>>>>>> of Matthias'
>>>>>>>>>> last reply. But given how this worked out it seems like we can
>>>>>> just focus
>>>>>>>>>> on adding
>>>>>>>>>> support for Headers in the DSL in this KIP by building off of the
>>>>>>>>>> groundwork of
>>>>>>>>>> KIP-478? It doesn't seem necessary to go back and add support for
>>>>>> headers
>>>>>>>>>> in the old
>>>>>>>>>> PAPI, since this will (or already has?) been deprecated.
>>>>>>>>>>
>>>>>>>>>> The one challenge is that this will presumably require that we
>>>>>> migrate
>>>>>>>>>> all
>>>>>>>>>> DSL operators
>>>>>>>>>> to the new PAPI before adding header support for those operators.
>>>>>> But
>>>>>>>>>> that
>>>>>>>>>> definitely
>>>>>>>>>> sounds achievable here
>>>>>>>>>>
>>>>>>>>>> On Wed, Oct 28, 2020 at 11:10 AM Jorge Esteban Quilcate Otoya <
>>>>>>>>>> quilcate.jorge@gmail.com> wrote:
>>>>>>>>>>
>>>>>>>>>>> Hi Matthias,
>>>>>>>>>>>
>>>>>>>>>>> Sorry for the late reply.
>>>>>>>>>>>
>>>>>>>>>>> I like the proposal. Just to check if I got it right:
>>>>>>>>>>>
>>>>>>>>>>> We can extend the `kstream.to()` function to support setting
>>>>>> headers.
>>>>>>>>>>> e.g.:
>>>>>>>>>>>
>>>>>>>>>>> ```
>>>>>>>>>>> void to(final String topic,
>>>>>>>>>>> final Produced<K, V> produced,
>>>>>>>>>>> final HeadersExtractor<K, V> headersExtractor);
>>>>>>>>>>> ```
>>>>>>>>>>>
>>>>>>>>>>> where `HeadersExtractor`:
>>>>>>>>>>>
>>>>>>>>>>> ```
>>>>>>>>>>> public interface HeadersExtractor<K, V> {
>>>>>>>>>>> Headers extract(final K key, final V value, final
>>>>>> RecordContext
>>>>>>>>>>> recordContext);
>>>>>>>>>>> }
>>>>>>>>>>> ```
>>>>>>>>>>>
>>>>>>>>>>> This would require to change `Topology#addSink()` to support
>>>>>> this
>>>>>>>>>>> extractor as well.
>>>>>>>>>>>
>>>>>>>>>>> If this is aligned with your proposal, I'm happy to add it to
>>>>>> this KIP.
>>>>>>>>>>>
>>>>>>>>>>> Cheers,
>>>>>>>>>>> Jorge.
>>>>>>>>>>>
>>>>>>>>>>> On Fri, Sep 11, 2020 at 11:03 PM Matthias J. Sax <
>>>>>> mjsax@apache.org>
>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Jorge,
>>>>>>>>>>>>
>>>>>>>>>>>> thanks a lot for this KIP. Being able to modify headers is a
>>>>>> very
>>>>>>>>>>>> valuable feature.
>>>>>>>>>>>>
>>>>>>>>>>>> However, before we actually expose them in the DSL, I am
>>>>>> wondering
>>>>>>>>>> if we
>>>>>>>>>>>> should improve how headers can be modified in the PAPI?
>>>>>> Currently,
>>>>>>>>>> it is
>>>>>>>>>>>> possible but very clumsy to work with headers in the
>>>>>> Processor API,
>>>>>>>>>>>> because of two reasons:
>>>>>>>>>>>>
>>>>>>>>>>>> (1) There is no default `Headers` implementation in the
>>>>>> public API
>>>>>>>>>>>> (2) There is no explicit way to set headers for output
>>>>>> records
>>>>>>>>>>>>
>>>>>>>>>>>> Currently, the input record headers are copied into the output
>>>>>>>>>> records
>>>>>>>>>>>> when `forward()` is called, however, it's not really a deep
>>>>>> copy but
>>>>>>>>>> we
>>>>>>>>>>>> just copy the reference. This implies that one needs to work
>>>>>> with a
>>>>>>>>>>>> single mutable object that flows through multiple processors
>>>>>> making
>>>>>>>>>> it
>>>>>>>>>>>> very error prone.
>>>>>>>>>>>>
>>>>>>>>>>>> Furthermore, if you want to emit multiple output records, and
>>>>>> for
>>>>>>>>>>>> example want to add two different headers to the output record
>>>>>>>>>> (based on
>>>>>>>>>>>> the same input headers), you would need to do something like
>>>>>> this:
>>>>>>>>>>>>
>>>>>>>>>>>> Headers h = context.headers();
>>>>>>>>>>>> h.add(...);
>>>>>>>>>>>> context.forward(...);
>>>>>>>>>>>> // remove the header you added for the first output record
>>>>>>>>>>>> h.remove(...);
>>>>>>>>>>>> h.add(...);
>>>>>>>>>>>> context.forward(...);
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> Maybe we could extend `To` to allow passing in a new
>>>>>> `Headers` object
>>>>>>>>>>>> (or an `Iterable<Header>` similar to `ProducerRecord`)? We
>>>>>> could
>>>>>>>>>> either
>>>>>>>>>>>> add it to your KIP or do a new KIP just for the PAPI.
>>>>>>>>>>>>
>>>>>>>>>>>> Thoughts?
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> -Matthias
>>>>>>>>>>>>
>>>>>>>>>>>> On 7/16/20 4:05 PM, Jorge Esteban Quilcate Otoya wrote:
>>>>>>>>>>>>> Hi everyone,
>>>>>>>>>>>>>
>>>>>>>>>>>>> Bumping this thread to check if there's any feedback.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>> Jorge.
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Tue, Jun 30, 2020 at 12:46 AM Jorge Esteban Quilcate
>>>>>> Otoya <
>>>>>>>>>>>>> quilcate.jorge@gmail.com> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> Hi everyone,
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> I would like to start the discussion for KIP-634:
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-634%3A+Complementary+support+for+headers+in+Kafka+Streams+DSL
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Looking forward to your feedback.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Thanks!
>>>>>>>>>>>>>> Jorge.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>
>>>>>>
>>>
>>
>