Warren, thanks for following up this KIP. And sorry for the "messy" discussion thread. Adding this feature is a little tricky. We still hope to get it into 1.2 release, but atm there is not much progress.
However, for your use case, you can replace .map() with .transform() that allows you to access the record's timestamp (via the provided `context` object) as extracted from the TimestampExtractor. See the docs for more details: https://kafka.apache.org/documentation/streams/developer-guide/dsl-api.html#applying-processors-and-transformers-processor-api-integration -Matthias On 3/13/18 12:51 PM, Warren, Brad wrote: > Hi devs, > > > > It’s a bit difficult to put all of the pieces together regarding the > status and API changes around the KIPs dealing with exposing the record > metadata in the Processor and DSL APIs. This is a feature that my team > here at American Airlines is keenly interested in and I’d like to > provide a real world use case to help move the discussion along: > > > > I have a source topic that contains a text value that includes datetimes > without a year. The desire is to order the records in a stream by an > extracted timestamp from the record value and we plan to use the > timestamp from the source topic to provide the year. We’re hoping to > use the DSL. Something like: > > > > val streamOrderedByMyValueTime = Builder.stream(“sourceTopic”).map( K,V > -> KeyValue(KR, VR, timestamp) ) > > > > so then I can do > > > > groupBy(), aggregate(), etc. > > > > Inside the mapper, my timestamp would be something like > LocalDateTime.of(yearFromIncomingConsumerRecordTimestamp, > monthFromValue, dayFromValue, ….) > > > > Looking at the wiki here > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=73637757, > what is the proposed implementation of RichValueMapper? Is it going to > support what I want to do here? > > > > > > Thanks, > > Brad > > > > cid:49F8CA06-65F7-457B-9DC0-8251F696295B > > > > *Brad Warren*** > > /Principal Application Architect/ > > /Airport Technology/// > > > > brad.war...@aa.com > > > > cid:DB82A805-2411-4411-8D3D-3688F7234324 > > > > > > > > > > NOTICE: This email and any attachments are for the exclusive and > confidential use of the intended recipient(s). If you are not an > intended recipient, please do not read, distribute, or take action in > reliance upon this message. If you have received this in error, please > notify me immediately by return email and promptly delete this message > and its attachments from your computer. >
signature.asc
Description: OpenPGP digital signature
