package Ġorg . xy ro . k umul us Ċ Ċ import Ġmu . Kotlin Logging Ċ import Ġorg . apache . storm . Config Ċ import Ġorg . apache . storm . sp out . Sp out Output Collector Ċ import Ġorg . apache . storm . task . Topology Context Ċ import Ġorg . apache . storm . topology . Basic Output Collector Ċ import Ġorg . apache . storm . topology . Output Fields Decl arer Ċ import Ġorg . apache . storm . topology . base . Base Basic B olt Ċ import Ġorg . apache . storm . tuple . Fields Ċ import Ġorg . apache . storm . tuple . Tuple Ċ import Ġorg . junit . Test Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ class ĠTest Single Ack ing Ġ{ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Single Ack Per Tree Topology () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuilder Ġ= Ġorg . apache . storm . topology . Topology Builder () ĊĠĠĠĠĠĠĠ Ġval Ġconfig : ĠMutable Map < String , ĠAny > Ġ= Ġmutable Map Of () ĊĊĠĠĠĠĠĠĠ Ġconfig [ Config . TO PO LOG Y _ MAX _ S PO UT _ PENDING ] Ġ= Ġ 5 L ĊĠĠĠĠĠĠĠ Ġconfig [ Config . TO PO LOG Y _ MESSAGE _ TIMEOUT _ SE CS ] Ġ= Ġ 5 L ĊĠĠĠĠĠĠĠ Ġconfig [ K umul us Topology . CONF _ THREAD _ POOL _ CORE _ SIZE ] Ġ= Ġ 5 L ĊĊĠĠĠĠĠĠĠ Ġbuilder . set Sp out (" sp out ", ĠTest Sp out ()) ĊĠĠĠĠĠĠĠ Ġbuilder . set B olt (" bolt 1 ", ĠTest B olt ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. none Grouping (" sp out ") ĊĠĠĠĠĠĠĠ Ġbuilder . set B olt (" bolt 2 ", ĠTest B olt ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. none Grouping (" sp out ") ĊĊĠĠĠĠĠĠĠ Ġval Ġst orm Topology Ġ= Ġbuilder . create Topology () !! ĊĠĠĠĠĠĠĠ Ġval Ġk umul us Topology Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK umul us Storm Transformer . initialize Topology ( storm Topology , Ġconfig , Ġ" test ") ĊĠĠĠĠĠĠĠ Ġk umul us Topology . prepare ( 1 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠĠĠĠĠ Ġk umul us Topology . start ( false ) ĊĠĠĠĠĠĠĠ ĠThread . sleep ( 5 0 0 0 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠTest Sp out Ġ: ĠDummy Sp out ({ Ġit . declare ( Fields (" id ")) Ġ}) Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġvar Ġindex : ĠInt Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġemitted Ġ= Ġmutable Set Of < Int >() ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġopen ( conf : ĠMutable Map < Any ?, ĠAny ?> ?, Ġcontext : ĠTop ology Context ?, Ġcollector : ĠSp out Output Collector ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . open ( conf , Ġcontext , Ġcollector ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . index Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġfail ( msg Id : ĠAny ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfinish ( msg Id ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġack ( msg Id : ĠAny ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfinish ( msg Id ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġfinish ( msg Id : ĠAny ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(( msg Id Ġas ĠInt ) Ġ% Ġ 1 0 0 _ 0 0 0 Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" Ack ing Ġ$ msg Id " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! emit ted . remove ( msg Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRuntimeException (" msg Id Ġ'$ msg Id ' Ġhad Ġalready Ġbeen Ġack ed ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġnext Tuple () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠmessageId Ġ= Ġ++ index ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemitted . add ( message Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcollector . emit ( listOf ( message Id ), ĠmessageId ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġclass ĠTest B olt Ġ: ĠBase Basic B olt () Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġexecute ( p 0 : ĠTuple ?, Ġp 1 : ĠBasic Output Collector ?) Ġ= ĠUnit ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġdeclare Output Fields ( p 0 : ĠOutput Fields Decl arer ?) Ġ= ĠUnit ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} ĊĠĠĠ Ġ} Ċ } Ċ