> 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 <vvcep...@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.jo...@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.jo...@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 < > sop...@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.jo...@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 < > mj...@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.jo...@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. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
