@ file : Suppress (" UN CHECK ED _ CAST ") Ċ Ċ package Ġkafka . flow . consumer Ċ Ċ import Ġbe . delta . flow . time . milliseconds Ċ import Ġbe . delta . flow . time . seconds Ċ import Ġinvoke And Throw Ċ import Ġjava . time . Duration Ċ import Ġjava . time . Instant Ċ import Ġkafka . flow . Topic Descriptor Ċ import Ġkafka . flow . consumer . processor . Buffered Kafka Writer Sink Ċ import Ġkafka . flow . consumer . processor . Grouping Processor Ċ import Ġkafka . flow . consumer . processor . Kafka Writer Sink Ċ import Ġkafka . flow . consumer . processor . Sink Ċ import Ġkafka . flow . consumer . processor . Topic Descriptor Deserializer Processor Ċ import Ġkafka . flow . consumer . processor . Transaction Processor Ċ import Ġkafka . flow . consumer . processor . Transform Processor Ċ import Ġkafka . flow . consumer . with . group . id . Maybe Transaction Ċ import Ġkafka . flow . consumer . with . group . id . With Transaction Ċ import Ġkafka . flow . consumer . with . group . id . Without Transaction Ċ import Ġkafka . flow . producer . Kafka Output Ċ import Ġkafka . flow . producer . Kafka Output Record Ċ import Ġkafka . flow . utils . Flow De bounc er Ċ import Ġkafka . flow . utils . Flow De bounc er . Companion . de bounce Ċ import Ġkotlinx . coroutines . Flow Preview Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . as Flow Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . flow . filter Ċ import Ġkotlinx . coroutines . flow . filter Is Instance Ċ import Ġkotlinx . coroutines . flow . flatMap Concat Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġkotlinx . coroutines . flow . map NotNull Ċ import Ġkotlinx . coroutines . flow . on Completion Ċ import Ġkotlinx . coroutines . flow . on Each Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠWithout Transaction >> . collect Values ( block : Ġsuspend Ġ( Value ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠ Ġthis . collect Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( it Ġis ĠRecord ) Ġblock . invoke And Throw ( it . value ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠWith Transaction >> . collect Values ( block : Ġsuspend Ġ( Value , ĠWith Transaction ) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠ Ġthis . collect Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( it Ġis ĠRecord ) Ġblock . invoke And Throw ( it . value , Ġit . transaction ) ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . collect Records ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( Record < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >) Ġ-> ĠUnit Ċ ) Ġ{ ĊĠĠĠ Ġthis . collect Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( it Ġis ĠRecord ) Ġblock . invoke And Throw ( it ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In Ġ: ĠMaybe Transaction , ĠKey Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In >> . map Record ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( Record < Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In >) Ġ-> ĠRecord < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out > Ċ ): ĠFlow < Kafka Message < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out >> Ġ{ ĊĠĠĠ Ġreturn Ġmap Ġ{ Ġkafka Message Ġ-> ĊĠĠĠĠĠĠĠ Ġwhen Ġ( kafka Message ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠRecord Ġ-> Ġblock . invoke And Throw ( kafka Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠFlow Control Message Ġ-> Ġkafka Message Ġas ĠKafka Message < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out > ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In Ġ: ĠMaybe Transaction , ĠKey Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In >> . map Record NotNull ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( Record < Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In >) Ġ-> ĠRecord < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out >? Ċ ): ĠFlow < Kafka Message < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out >> Ġ{ ĊĠĠĠ Ġreturn Ġmap NotNull Ġ{ Ġkafka Message Ġ-> ĊĠĠĠĠĠĠĠ Ġwhen Ġ( kafka Message ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠRecord Ġ-> Ġblock . invoke And Throw ( kafka Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠFlow Control Message Ġ-> Ġkafka Message Ġas ĠKafka Message < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out > ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠUnit , ĠOutput , ĠTransaction >> . deserialize Value ( block : Ġsuspend Ġ( ByteArray ) Ġ-> ĠValue ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġmap Record Ġ{ Ġrecord Ġ-> ĊĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . consumer Record , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . key , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . partition Key , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( record . consumer Record . value ()), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . timestamp , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . output , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . transaction ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠWithout Transaction >> . values (): ĠFlow < Value > Ġ{ ĊĠĠĠ Ġreturn Ġfilter Is Instance < Record < Key , ĠPartition , ĠValue , ĠOutput , ĠWithout Transaction >>() ĊĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . value Ġ} Ċ } Ċ Ċ @ Jvm Name (" values Key Partition Value Output With Transaction ") Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠWith Transaction >> . values (): ĠFlow < Pair < Key , ĠValue >> Ġ{ ĊĠĠĠ Ġreturn Ġfilter Is Instance < Record < Key , ĠPartition , ĠValue , ĠOutput , ĠWith Transaction >>() ĊĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠPair ( it . key , Ġit . value ) Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on Each Record ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( Record < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >) Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠRecord ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( message ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . onStart Consum ing ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( client : ĠKafka Flow Consumer < Flow < Kafka Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >> >) Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠStart Consum ing ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( message . client ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on Stop Consum ing ( ĊĠĠĠ Ġblock : Ġsuspend Ġ() Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠStop Consum ing ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw () ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on EndOf Batch ( ĊĠĠĠ Ġblock : Ġsuspend Ġ() Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠEnd Of Batch ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw () ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on Partition Assigned ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( new ly Assigned Partition : ĠList < Topic Partition >, Ġnew Assignment : ĠList < Topic Partition >) Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠPart itions Assigned ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( message . new ly Assigned Partitions , Ġmessage . new Assignment ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on Partition Rev oked ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( rev oked Partition : ĠList < Topic Partition >, Ġnew Assignment : ĠList < Topic Partition >) Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠPart itions Rev oked ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( message . rev oked Partitions , Ġmessage . new Assignment ) ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on Partition Changed ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( new Assignment : ĠList < Topic Partition >) Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠPartition Changed Message ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( message . new Assignment ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Unit , ĠUnit , ĠUnit , ĠOutput , ĠTransaction >> . deserialize Using ( ĊĠĠĠ Ġtopic Descriptor : ĠTopic Descriptor < Key , ĠPartition , ĠValue >, ĊĠĠĠ Ġon Deserialization Exception : Ġsuspend Ġ( Throwable ) Ġ-> ĠUnit Ġ= Ġ{ Ġit . printStackTrace () Ġ} Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue ?, ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġdeserialize Using ( listOf ( topic Descriptor ), Ġon Deserialization Exception ) Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Unit , ĠUnit , ĠUnit , ĠOutput , ĠTransaction >> . deserialize Using ( ĊĠĠĠ Ġtopic Descriptors : ĠList < Topic Descriptor < out ĠKey , Ġout ĠPartition , Ġout ĠValue >>, ĊĠĠĠ Ġon Deserialization Exception : Ġsuspend Ġ( Throwable ) Ġ-> ĠUnit Ġ= Ġ{ Ġit . printStackTrace () Ġ} Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue ?, ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġtransform ( Topic Descriptor Deserializer Processor ( topic Descriptors Ġas ĠList < Topic Descriptor < Key , ĠPartition , ĠValue >>, Ġon Deserialization Exception )) Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue ?, ĠOutput , ĠTransaction >> . ignore T omb stones (): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġfilter Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠRecord ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġmessage . value Ġ!= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! result ) Ġmessage . transaction . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġas ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In Ġ: ĠMaybe Transaction , ĠKey Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In >> . transform ( ĊĠĠĠ Ġprocessor : ĠTransform Processor < Key In , ĠPartition In , ĠValue In , ĠOutput In , ĠTransaction In , ĠKey Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out > Ċ ): ĠFlow < Kafka Message < Key Out , ĠPartition Out , ĠValue Out , ĠOutput Out , ĠTransaction Out >> Ġ{ ĊĠĠĠ Ġreturn Ġthis ĊĠĠĠĠĠĠĠ Ġ. onStart Consum ing ( processor :: start Consum ing ) ĊĠĠĠĠĠĠĠ Ġ. on Stop Consum ing ( processor :: stop Consum ing ) ĊĠĠĠĠĠĠĠ Ġ. on Completion Ġ{ Ġprocessor . completion () Ġ} ĊĠĠĠĠĠĠĠ Ġ. on EndOf Batch ( processor :: end Of Batch ) ĊĠĠĠĠĠĠĠ Ġ. on Partition Assigned ( processor :: partition Assigned ) ĊĠĠĠĠĠĠĠ Ġ. on Partition Rev oked ( processor :: partition Rev oked ) ĊĠĠĠĠĠĠĠ Ġ. map Record NotNull Ġ{ Ġprocessor . record ( it . consumer Record , Ġit . key , Ġit . partition Key , Ġit . value , Ġit . timestamp , Ġit . output , Ġit . transaction ) Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . collect ( ĊĠĠĠ Ġprocessor : ĠS ink < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction > Ċ ) Ġ{ ĊĠĠĠ Ġreturn Ġthis ĊĠĠĠĠĠĠĠ Ġ. onStart Consum ing ( processor :: start Consum ing ) ĊĠĠĠĠĠĠĠ Ġ. on Stop Consum ing ( processor :: stop Consum ing ) ĊĠĠĠĠĠĠĠ Ġ. on Completion Ġ{ Ġprocessor . completion () Ġ} ĊĠĠĠĠĠĠĠ Ġ. on EndOf Batch ( processor :: end Of Batch ) ĊĠĠĠĠĠĠĠ Ġ. on Partition Assigned ( processor :: partition Assigned ) ĊĠĠĠĠĠĠĠ Ġ. on Partition Rev oked ( processor :: partition Rev oked ) ĊĠĠĠĠĠĠĠ Ġ. on Each Record Ġ{ Ġprocessor . record ( it . consumer Record , Ġit . key , Ġit . partition Key , Ġit . value , Ġit . timestamp , Ġit . output , Ġit . transaction ) Ġ} ĊĠĠĠĠĠĠĠ Ġ. collect () Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠWithout Transaction >> . create Transactions ( ĊĠĠĠ Ġmax Open Transactions : ĠInt Ġ= Ġ 1 0 2 4 , ĊĠĠĠ Ġcommit Interval : ĠDuration Ġ= Ġ 3 0 . seconds () Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠWith Transaction >> Ġ{ ĊĠĠĠ Ġreturn Ġthis . transform ( Transaction Processor ( max Open Transactions , Ġcommit Interval )) Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠKafka Output , ĠWith Transaction >> . write Output To Kafka And Commit () Ġ{ ĊĠĠĠ Ġthis . collect ( Kafka Writer Sink ()) Ċ } Ċ Ċ @ Jvm Name (" write Output To Kafka And Commit Key Partition Value Unit With Transaction Kafka Output ") Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue > ĠFlow < Pair < List < Kafka Message < Key , ĠPartition , ĠValue , ĠUnit , ĠWith Transaction >>, ĠKafka Output >> . write Output To Kafka And Commit () Ġ{ ĊĠĠĠ Ġval Ġkafka Writer Ġ= ĠBuffered Kafka Writer Sink < Key , ĠPartition , ĠValue , ĠWith Transaction >() ĊĠĠĠ Ġthis . collect Ġ{ Ġkafka Writer . handle Records ( it ) Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠKafka Output , ĠWithout Transaction >> . write Output To Kafka () Ġ{ ĊĠĠĠ Ġthis . collect ( Kafka Writer Sink ()) Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠUnit , ĠTransaction >> . map Value To Output ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( Key , ĠValue ) Ġ-> ĠOutput Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġmap Record Ġ{ Ġrecord Ġ-> ĠRecord ( record . consumer Record , Ġrecord . key , Ġrecord . partition Key , Ġrecord . value , Ġrecord . timestamp , Ġblock . invoke And Throw ( record . key , Ġrecord . value ), Ġrecord . transaction ) Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . groupBy Partition Key ( ĊĠĠĠ Ġprocessor Timeout : ĠDuration , ĊĠĠĠ Ġchannel Capacity : ĠInt Ġ= Ġ 1 0 , ĊĠĠĠ Ġflow Factory : Ġsuspend Ġ( Flow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >>, Ġpartition Key : ĠPartition ) Ġ-> ĠUnit Ċ ) Ġ{ ĊĠĠĠ Ġreturn Ġcollect ( Grouping Processor ( processor Timeout , Ġchannel Capacity , Ġflow Factory )) Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . filter Value ( ĊĠĠĠ Ġpredicate : Ġsuspend Ġ( Value ) Ġ-> ĠBoolean Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġfilter Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠRecord ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġpredicate . invoke And Throw ( message . value ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! result ) Ġmessage . transaction . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . on Each T omb stone ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( Record < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >) Ġ-> ĠUnit Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġon Each Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠRecord Ġ&& Ġmessage . value Ġ== Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock . invoke And Throw ( message ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < List < Kafka Message < Key , ĠPartition , ĠValue , ĠUnit , ĠTransaction >> >. map Values To Output ( ĊĠĠĠ Ġblock : Ġsuspend Ġ( List < Pair < Key , ĠValue >>) Ġ-> ĠOutput Ċ ): ĠFlow < Pair < List < Kafka Message < Key , ĠPartition , ĠValue , ĠUnit , ĠTransaction >>, ĠOutput >> Ġ{ ĊĠĠĠ Ġreturn Ġmap Ġ{ Ġrecords Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġkey Values Ġ= Ġrecords ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Is Instance < Record < Key , ĠPartition , ĠValue , ĠUnit , ĠTransaction >>() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . key Ġto Ġit . value Ġ} ĊĠĠĠĠĠĠĠ Ġrecords Ġto Ġblock . invoke And Throw ( key Values ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> . de bounce Input On Key ( ĊĠĠĠ Ġtime Provider : Ġ( Record < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >, ĠInstant ?) Ġ-> ĠInstant ?, ĊĠĠĠ Ġmax De bounce Duration : ĠDuration , ĊĠĠĠ Ġinterval : ĠDuration Ġ= Ġ 1 0 . milliseconds (), ĊĠĠĠ Ġclean Up Interval : ĠDuration Ġ= Ġ 1 0 . seconds (), Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġdebounce ( ĊĠĠĠĠĠĠĠ Ġ{ Ġmessage Ġ-> Ġif Ġ( message Ġis ĠRecord < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >) ĠPair ( message . consumer Record . topic (), Ġmessage . key ) Ġelse Ġnull Ġ}, ĊĠĠĠĠĠĠĠ Ġ{ Ġmessage , Ġinstant Ġ-> Ġif Ġ( message Ġis ĠRecord < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >) Ġtime Provider . invoke And Throw ( message , Ġinstant ) Ġelse Ġnull Ġ}, ĊĠĠĠĠĠĠĠ Ġmax De bounce Duration , ĊĠĠĠĠĠĠĠ Ġinterval , ĊĠĠĠĠĠĠĠ Ġclean Up Interval ĊĠĠĠ Ġ). filter Skip Actions And Commit () Ċ } Ċ Ċ private Ġfun Ġ< Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Flow De bounc er . Action < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> >. filter Skip Actions And Commit (): ĊĠĠĠĠĠĠĠ ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġmap NotNull Ġ{ Ġaction Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( action Ġis ĠFlow De bounc er . Skip <* >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( action . data Ġas ĠRecord < Key , ĠPartition , ĠValue , ĠOutput , ĠTransaction >). transaction . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction . data ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ @ Flow Preview Ċ public Ġsuspend Ġfun Ġ< Key , ĠPartition , ĠValue , ĠTransaction Ġ: ĠMaybe Transaction > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠKafka Output , ĠTransaction >> . de bounce Output On Key ( ĊĠĠĠ Ġtime Provider : Ġ( Kafka Output Record , ĠInstant ?) Ġ-> ĠInstant ?, ĊĠĠĠ Ġmax De bounce Duration : ĠDuration , ĊĠĠĠ Ġinterval : ĠDuration Ġ= Ġ 1 0 . milliseconds (), ĊĠĠĠ Ġclean Up Interval : ĠDuration Ġ= Ġ 1 0 . seconds (), Ċ ): ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠKafka Output , ĠTransaction >> Ġ{ ĊĠĠĠ Ġreturn Ġsplit Multiple Output To Single Messages () ĊĠĠĠĠĠĠĠ Ġ. de bounce ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġmessage Ġ-> Ġmessage . get Single Output Record OrNull ()?. let Ġ{ ĠPair ( it . topic Descriptor . name , Ġit . key ) Ġ} Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġmessage , Ġinstant Ġ-> Ġmessage . get Single Output Record OrNull ()?. let Ġ{ Ġtime Provider . invoke And Throw ( it , Ġinstant ) Ġ} Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmax De bounce Duration , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinterval , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclean Up Interval ĊĠĠĠĠĠĠĠ Ġ). filter Skip Actions And Commit () Ċ } Ċ Ċ @ Flow Preview Ċ private Ġfun Ġ< Key , ĠPartition , ĠTransaction Ġ: ĠMaybe Transaction , ĠValue > ĠFlow < Kafka Message < Key , ĠPartition , ĠValue , ĠKafka Output , ĠTransaction >> . split Multiple Output To Single Messages () Ġ= ĊĠĠĠ Ġflat Map Concat Ġ{ Ġmessage Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( message Ġis ĠRecord < Key , ĠPartition , ĠValue , ĠKafka Output , ĠTransaction > Ġ&& Ġmessage . output . records . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= Ġmessage . output . records . map Ġ{ Ġoutput Record Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage . transaction . lock () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( message . consumer Record , Ġmessage . key , Ġmessage . partition Key , Ġmessage . value , Ġmessage . timestamp , ĠKafka Output ( listOf ( output Record )), Ġmessage . transaction ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage . transaction . unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessages . as Flow () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( message ). as Flow () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ private Ġfun Ġ< Key , ĠPartition , ĠValue , ĠTransaction Ġ: ĠMaybe Transaction > ĠKafka Message < Key , ĠPartition , ĠValue , ĠKafka Output , ĠTransaction >. get Single Output Record OrNull (): ĠKafka Output Record ? Ġ{ ĊĠĠĠ Ġreturn Ġif Ġ( this Ġis ĠRecord Ġ&& Ġoutput . records . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire ( output . records . size Ġ== Ġ 1 ) Ġ{ Ġ" This Ġshould Ġonly Ġbe Ġused Ġif Ġthere Ġis Ġat Ġmax Ġ 1 Ġrecord Ġin Ġthe Ġoutput , Ġplease Ġuse Ġsplit Multiple Output To Single Messages Ġbefore Ġthis Ġfunction " Ġ} ĊĠĠĠĠĠĠĠ Ġoutput . records . first () ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġnull ĊĠĠĠ Ġ} Ċ }