package Ġio . z up . springframework . kafka . listener Ċ Ċ import Ġio . z up . springframework . kafka . helper . Test Constants Ċ import Ġkotlinx . coroutines . experimental . channels . Channel Ċ import Ġkotlinx . coroutines . experimental . run Blocking Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . springframework . kafka . annotation . Kafka Listener Ċ import Ġjava . util . concurrent . Count DownLatch Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ type alias ĠHandler < K , ĠV > Ġ= Ġ( Consumer Record < K , ĠV >) Ġ-> ĠUnit Ċ Ċ class ĠReceiver < K , ĠV > Ġ{ ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġl atch : ĠCount DownLatch ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġretry Channel : ĠChannel < Consumer Record < K , ĠV >> ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġdl q Channel : ĠChannel < Consumer Record < K , ĠV >> ĊĠĠĠ Ġprivate Ġvar Ġmain Handler : ĠHandler < K , ĠV >? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġretry Handler : ĠHandler < K , ĠV >? Ġ= Ġnull ĊĊĠĠĠ Ġfun Ġreset (): ĠReceiver < K , ĠV > Ġ= ĊĠĠĠĠĠĠĠ Ġthis . apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġl atch Ġ= ĠCount DownLatch ( 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretry Channel Ġ= ĠChannel ( Test Constants . MAX _ RE TR IES ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdl q Channel Ġ= ĠChannel ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmain Handler Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretry Handler Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwith Interaction Count ( count : ĠInt ): ĠReceiver < K , ĠV > Ġ= ĊĠĠĠĠĠĠĠ Ġthis . apply Ġ{ Ġl atch Ġ= ĠCount DownLatch ( count ) Ġ} ĊĊĠĠĠ Ġfun Ġwith MessageHandler ( handler : ĠHandler < K , ĠV >): ĠReceiver < K , ĠV > Ġ= ĊĠĠĠĠĠĠĠ Ġthis . apply Ġ{ Ġmain Handler Ġ= Ġhandler Ġ} ĊĊĠĠĠ Ġfun Ġwith Message Retry Handler ( handler : ĠHandler < K , ĠV >): ĠReceiver < K , ĠV > Ġ= ĊĠĠĠĠĠĠĠ Ġthis . apply Ġ{ Ġretry Handler Ġ= Ġhandler Ġ} ĊĊĠĠĠ Ġ@ Kafka Listener ( topics Ġ= Ġ[ Test Constants . MAIN _ TOPIC ], ĠerrorHandler Ġ= Ġ" error Handler ") ĊĠĠĠ Ġfun Ġhandle ( record : ĠConsumer Record < K , ĠV >) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmain Handler ?. invoke ( record ) ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Kafka Listener ( topics Ġ= Ġ[ Test Constants . RETRY _ TOPIC ], ĠerrorHandler Ġ= Ġ" error Handler ") ĊĠĠĠ Ġfun Ġhandle Retry ( record : ĠConsumer Record < K , ĠV >) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretry Handler ?. invoke ( record ) ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretry Channel . send ( record ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Kafka Listener ( topics Ġ= Ġ[ Test Constants . DL Q _ TOPIC ]) ĊĠĠĠ Ġfun Ġhandle D lq ( record : ĠConsumer Record < K , ĠV >) Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdl q Channel . send ( record ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġl atch . count Down () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġawait Retry (): ĠConsumer Record < K , ĠV > Ġ= ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretry Channel . receive () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġawait Dead Letter (): ĠConsumer Record < K , ĠV > Ġ= ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdl q Channel . receive () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġawait () Ġ= ĊĠĠĠĠĠĠĠ Ġl atch . await ( 5 , ĠTimeUnit . SECONDS ) Ċ Ċ }