Hi Theo, thanks for reaching out to the community. I am pulling in Aljoscha and Klou who have worked on the new WatermarkStrategy/WatermarkGenerator abstraction and might be able to help you with your problem. At the moment, it looks to me that there is no way to combine state with the new WatermarkGenerator abstraction.
Cheers, Till On Wed, Aug 19, 2020 at 3:37 PM Theo Diefenthal < theo.diefent...@scoop-software.de> wrote: > Hi there, > > Right now I'm in the process of upgrading our Flink 1.9 jobs to Flink 1.11. > > In Flink 1.9, I was able to write a AssignerWithperiodicWatermarks which > also extended AbstractRichFunction and could thus utilize State and > getRuntimeContext() in there. This worked as the > TimestampsAndWatermarksOperator was a AbstractUdfStreamOperator and passed > my assigner in as the userFunction to that operator. > > I used this feature for some "per partition processing" which Flinks > somehow isn't ideally suited for at the moment I guess. We have ascending > watermarks per kafka partition and do some processing on that. In order to > maintain state per kafka-partition, I now keyby kafkapartition in our > stream (not ideal but better than operatorstate in terms of rescaling) but > afterwards need to emulate the watermark strategy from the initial kafka > source, i.e. reassign watermarks the same way as the kafka source did (per > kafka partition within the operator). Via getRuntimeContext() I am/was able > to identify the kafkaPartitions one operatorinstance was responsible for > and could produce the outputwatermark accordingly. (min over all > responsible partitions). > > In Flink 1.11, how can I rebuild this behavior? Do I really need to build > my own TimestampsAndWatermarksOperator which works like the old one? Or is > there a better approach? > > Best regards > Theo >
