package Ġno . nav . hel se . streams Ċ Ċ import Ġio . con fluent . kafka . serializers . Abstract Kafka Av ro Ser De Config Ċ import Ġorg . apache . kafka . common . serialization . Ser des Ċ import Ġorg . apache . kafka . streams . Streams Builder Ċ import Ġorg . apache . kafka . streams . k stream . Cons umed Ċ import Ġorg . apache . kafka . streams . k stream . K Stream Ċ import Ġorg . apache . kafka . streams . k stream . Pro duced Ċ Ċ private Ġval Ġstrings Ġ= ĠSer des . String () Ċ private Ġval Ġjson Ġ= ĠSer des . serde From ( Json Serializer (), ĠJson Deserializer ()) Ċ Ċ object ĠTop ics Ġ{ ĊĊĠĠ Ġ/** ĊĠĠĠ Ġ* ĠLegacy - input Ġto Ġthe Ġwhole Ġprocess . ĠContains Ġthe Ġoriginal Ġapplication . ĊĠĠĠ Ġ*/ ĊĠĠ Ġval ĠSY KE PEN GES Ã ĺ KN ADER _ IN N _ LEG ACY Ġ= ĠTopic ( ĊĠĠ Ġname Ġ= Ġ" sy fo - so kn ad - v 1 ", ĊĠĠ Ġkey Ser de Ġ= Ġstrings , ĊĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) ĊĊĠĠ Ġ/** ĊĠĠĠ Ġ* ĠInput Ġto Ġthe Ġwhole Ġprocess . ĠContains Ġthe Ġoriginal Ġapplication . ĊĠĠĠ Ġ*/ ĊĠĠ Ġval ĠSY KE PEN GES Ã ĺ KN ADER _ IN N Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" sy fo - so kn ad - v 2 ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= Ġstrings , ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) ĊĊĠĠ Ġ/** ĊĠĠĠ Ġ* ĠPro duced Ġby Ġ` split t ` Ġand Ġ` spin ne `, Ġ( at Ġleast ) Ġconsumed Ġby Ġ` spin ne `. ĊĠĠĠ Ġ* ĊĠĠĠ Ġ* ĠContains Ġapplications Ġen ric hed Ġwith Ġsupporting Ġinformation . ĊĠĠĠ Ġ*/ ĊĠĠ Ġval ĠSY KE PEN GE BE HAND LING Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" priv at - sy ke pen ge be handling ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= ĠSer des . String (), ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) ĊĊĠĠ Ġval ĠV ED TA K _ INF OT RY GD Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" priv at - hel se - info try gd - ved tak ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= Ġstrings , ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) ĊĊĠĠ Ġval ĠV ED TA K _ SY KE PEN GER Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" a ap en - hel se - sy ke p enger - ved tak ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= Ġstrings , ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) ĊĠĠ Ġval ĠV ED TA K _ KO MB IN ERT Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" priv at - hel se - ved tak - kom bin ert ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= Ġstrings , ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) ĊĊĠĠ Ġval ĠV ED TA K _ RESULT AT Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" priv at - hel se - ved tak - result at ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= Ġstrings , ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġstrings ĊĠĠ Ġ) ĊĊĠĠ Ġval ĠSY KE PEN GE BE HAND L INGS FE IL Ġ= ĠTopic ( ĊĠĠĠĠĠ Ġname Ġ= Ġ" priv at - hel se - sy ke p enger - be hand lings fe il ", ĊĠĠĠĠĠ Ġkey Ser de Ġ= ĠSer des . String (), ĊĠĠĠĠĠ Ġvalue Ser de Ġ= Ġjson ĊĠĠ Ġ) Ċ Ċ } Ċ Ċ fun Ġ< K Ġ: ĠAny , ĠV Ġ: ĠAny > ĠStreams Builder . consume Topic ( topic : ĠTopic < K , ĠV >): ĠK Stream < K , ĠV > Ġ{ ĊĠĠ Ġreturn Ġconsume Topic ( topic , Ġnull ) Ċ } Ċ Ċ fun Ġ< K : ĠAny , ĠV : ĠAny > ĠStreams Builder . consume Topic ( topic : ĠTopic < K , ĠV >, Ġschema Registry Url : ĠString ?): ĠK Stream < K , ĠV > Ġ{ ĊĠĠ Ġschema Registry Url ?. let Ġ{ ĊĠĠĠĠĠ Ġtopic . key Ser de . configure ( map Of ( ĊĠĠĠĠĠĠĠĠ ĠAbstract Kafka Av ro Ser De Config . SCHEMA _ REGISTRY _ URL _ CONFIG Ġto Ġschema Registry Url ĊĠĠĠĠĠ Ġ), Ġtrue ) ĊĊĠĠĠĠĠ Ġtopic . value Ser de . configure ( map Of ( ĊĠĠĠĠĠĠĠĠ ĠAbstract Kafka Av ro Ser De Config . SCHEMA _ REGISTRY _ URL _ CONFIG Ġto Ġschema Registry Url ĊĠĠĠĠĠ Ġ), Ġfalse ) ĊĠĠ Ġ} ĊĊĠĠ Ġreturn Ġstream < K , ĠV >( ĊĠĠĠĠĠ Ġtopic . name , ĠCon su med . with ( topic . key Ser de , Ġtopic . value Ser de ) ĊĠĠ Ġ) Ċ } Ċ Ċ fun Ġ< K , ĠV > ĠK Stream < K , ĠV >. to Topic ( topic : ĠTopic < K , ĠV >) Ġ{ ĊĠĠ Ġreturn Ġto ( topic . name , ĠPro duced . with ( topic . key Ser de , Ġtopic . value Ser de )) Ċ } Ċ