package Ġmicro . apps . service Ċ Ċ import Ġcom . github . java faker . F aker Ċ import Ġmicro . apps . k stream . K Ser des . grouped Ċ import Ġmicro . apps . k stream . K Ser des . pro duced With Ċ import Ġmu . Kotlin Logging Ċ import Ġorg . apache . kafka . streams . KeyValue Ċ import Ġorg . apache . kafka . streams . k stream . K Stream Ċ import Ġorg . apache . kafka . streams . state . Queryable Store Types . key Value Store Ċ import Ġorg . springframework . boot . autoconfigure . SpringBoot Application Ċ import Ġorg . springframework . boot . run Application Ċ import Ġorg . springframework . cloud . stream . binder . kafka . streams . Interactive Query Service Ċ import Ġorg . springframework . context . annotation . Bean Ċ import Ġorg . springframework . integration . support . Message Builder Ċ import Ġorg . springframework . kafka . support . Kafka Headers Ċ import Ġorg . springframework . messaging . Message Ċ import Ġorg . springframework . web . bind . annotation . GetMapping Ċ import Ġorg . springframework . web . bind . annotation . PathVariable Ċ import Ġorg . springframework . web . bind . annotation . RestController Ċ import Ġjava . util . Locale Ċ import Ġjava . util . UUID Ċ import Ġjava . util . function . Consumer Ċ import Ġjava . util . function . Function Ċ import Ġjava . util . function . Supplier Ċ import Ġorg . apache . kafka . streams . k stream . Material ized .` as ` Ġas Ġmaterial ized Ċ Ċ private Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} Ċ Ċ @ SpringBoot Application Ċ class ĠWord count Application Ġ{ ĊĊĠĠĠ Ġval Ġlocale Ġ= ĠLocale . getDefault () ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġproduce Ch uck Nor ris (): ĠSupplier < Message < String >> Ġ= ĠSupplier Ġ{ ĊĠĠĠĠĠĠĠ ĠMessage Builder . with Payload ( F aker . instance (). ch uck Nor ris (). fact ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. setHeader ( Kafka Headers . MESSAGE _ KEY , ĠUUID . randomUUID (). toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġconsume Counts (): ĠConsumer < K Stream < String , ĠLong >> Ġ= ĠConsumer Ġ{ ĊĠĠĠĠĠĠĠ Ġit . peek Ġ{ Ġk , Ġv Ġ-> Ġlogger . info (" COUNT : Ġ{} Ġ= Ġ{}", Ġk , Ġv ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġprocess Words (): ĠFunction < K Stream < String ?, ĠString >, ĠK Stream < String , ĠLong >> Ġ= ĠFunction Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcounts Stream Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Values Ġ{ Ġvalue : ĠString Ġ-> Ġvalue . lowercase ( locale ). split ("\\ W +". to Regex ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ_, Ġv Ġ-> Ġv . isNotBlank () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġ_: ĠString ?, Ġvalue : ĠString Ġ-> ĠKey Value ( value , Ġvalue ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. group ByKey ( grouped < String , ĠString >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ. window edBy ( Time Windows . of Size With No Grace ( of Minutes ( 5 )). advance By ( of Minutes ( 1 ))) Ġ// Ġ// Ġhop ping - windows ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. count ( material ized (" word - count - state - store ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Stream () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. peek Ġ{ Ġk , Ġv Ġ-> Ġlogger . info (" PROC ES SES ĠCOUNT S : Ġ{}, Ġ{}", Ġk , Ġv ) Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġ. peek Ġ{ Ġk , Ġv Ġ-> Ġlogger . info (" PROC ES SES ĠCOUNT S : Ġ{}, Ġ{}, Ġ{}, Ġ{}", Ġk . key (), Ġv , Ġk . window (). start (), Ġk . window (). end ()) Ġ} ĊĠĠĠĠĠĠĠ Ġ// Ġ. map Ġ{ Ġk , Ġv Ġ-> ĠKey Value ( k . key (), Ġv ) Ġ} Ġ// Ġunwrap Ġkey : ĠWindow ed < String > ĊĊĠĠĠĠĠĠĠ Ġcounts Stream . to (" counts ", Ġproduced With < String , ĠLong >()) ĊĠĠĠĠĠĠĠ Ġcounts Stream ĊĠĠĠ Ġ} Ċ } Ċ Ċ @ RestController Ċ class ĠIQ RestController ( private Ġval Ġi q Service : ĠInteractive Query Service ) Ġ{ ĊĠĠĠ Ġ@ GetMapping ("/ iq / count /{ word }") ĊĠĠĠ Ġfun Ġget Count (@ PathVariable Ġword : ĠString ): ĠLong Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstore Ġ= Ġi q Service . getQuery able Store (" word - count - state - store ", Ġkey Value Store < String , ĠLong >()) ĊĠĠĠĠĠĠĠ Ġreturn Ġstore [ word ] ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠ Ġrun Application < Word count Application >(* args ) Ċ } Ċ