package Ġch . max ant . k dc . mf . library Ċ Ċ import Ġch . max ant . k dc . mf . library . Context . Companion . REQUEST _ ID Ċ import Ġch . max ant . k dc . mf . library . Context . Companion . SESSION _ ID Ċ import Ġcom . fasterxml . jackson . core . Json ParseException Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġcom . n ha ar man . mockito _ kotlin .* Ċ import Ġio . op entr acing . Tracer Ċ import Ġio . op entr acing . noop . No op Tracer Factory Ċ import Ġio . small ry e . reactive . messaging . kafka . Incoming Kafka Record Metadata Ċ import Ġorg . apache . kafka . common . header . internals . Record Headers Ċ import Ġorg . eclipse . micro profile . reactive . messaging . Message Ċ import Ġorg . eclipse . micro profile . reactive . messaging . Metadata Ċ import Ġorg . jboss . logging . Logger Ċ import Ġorg . junit . jupiter . api . After Each Ċ import Ġorg . junit . jupiter . api . Assertions . assertEquals Ċ import Ġorg . junit . jupiter . api . Assertions . assertNull Ċ import Ġorg . junit . jupiter . api . BeforeEach Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . assert Throws Ċ import Ġjava . util . concurrent . Comp letableFuture Ċ import Ġjava . util . concurrent . Comp letableFuture . completed Future Ċ import Ġjava . util . concurrent . Comp letableFuture . failed Future Ċ import Ġjava . util . concurrent . Completion Stage Ċ import Ġjavax . interceptor . Invocation Context Ċ Ċ class ĠP imp ed And With D lt And Ack Test Ġ{ ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġlog : ĠLogger ĊĠĠĠ Ġprivate Ġlateinit Ġvar ĠerrorHandler : ĠError Handler ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġic : ĠInvocation Context ĊĠĠĠ Ġprivate Ġval Ġtracer : ĠTr acer Ġ= ĠNo op Tracer Factory . create () ĊĊĠĠĠ Ġ@ BeforeEach ĊĠĠĠ Ġfun Ġsetup () Ġ{ ĊĠĠĠĠĠĠĠ Ġlog Ġ= Ġmock {} ĊĠĠĠĠĠĠĠ ĠerrorHandler Ġ= Ġmock {} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ After Each ĊĠĠĠ Ġfun Ġteardown () Ġ{ ĊĠĠĠĠĠĠĠ Ġverify No More Interactions ( ic , Ġlog , ĠerrorHandler ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġannotation Added To Method With Wrong Num Params () Ġ{ ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Consumer Wrong Num Params :: class . java . getMethod (" process ")) Ġ// Ġ< --- ĠWR ONG ĠNUM ĠPARAM S ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Consumer Wrong Num Params ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen Ġ+ Ġthen ĊĠĠĠĠĠĠĠ ĠassertEquals (" EH 0 0 1 Ġ@ Error Handled Ġon Ġmethod Ġch . max ant . k dc . mf . library . Test Consumer Wrong Num Params :: process Ġmust Ġcontain Ġexactly Ġone Ġparameter ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThrows < IllegalArgumentException > Ġ{ ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ). invoke ( ic ) Ġ}. message ) ĊĠĠĠĠĠĠĠ Ġverify ( ic ). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). target ĊĠĠĠĠĠĠĠ Ġverify ( ic ). method ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġannotation Added To Method With Wrong Param Type () Ġ{ ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Consumer Wrong Param Type :: class . java . getMethod (" process ", ĠInt :: class . java )) Ġ// Ġ< --- ĠWR ONG ĠPARAM ĠTYPE ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Consumer Wrong Param Type ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen Ġ+ Ġthen ĊĠĠĠĠĠĠĠ ĠassertEquals (" EH 0 0 2 Ġ@ Error Handled Ġon Ġmethod Ġch . max ant . k dc . mf . library . Test Consumer Wrong Param Type :: process Ġmust Ġcontain Ġone ĠString / Message Ġparameter ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThrows < IllegalArgumentException > Ġ{ ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ). invoke ( ic ) Ġ}. message ) ĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġtimes ( 2 )). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). target ĊĠĠĠĠĠĠĠ Ġverify ( ic ). method ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġinvoke _ happy _ string _ not _ json () Ġ{ ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of (" asdf ")) Ġ// Ġ< --- ĠNOT ĠJSON ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test String Consumer Send To D lt :: class . java . getMethod (" process ", ĠString :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test String Consumer Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen Ġ+ Ġthen ĊĠĠĠĠĠĠĠ ĠassertEquals (""" Unrecognized Ġtoken Ġ' asdf ': Ġwas Ġexpecting Ġ( JSON ĠString , ĠNumber , ĠArray , ĠObject Ġor Ġtoken Ġ' null ', Ġ' true ' Ġor Ġ' false ') Ċ Ġat Ġ[ Source : Ġ( String )" asdf "; Ġline : Ġ 1 , Ġcolumn : Ġ 5 ] """, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThrows < Json ParseException > Ġ{ Ġsut . invoke ( ic ) Ġ}. message ) ĊĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġtimes ( 2 )). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġnever ()). proceed () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġinvoke _ happy _ string () Ġ{ ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ("{} ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test String Consumer Send To D lt :: class . java . getMethod (" process ", ĠString :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test String Consumer Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġwhenever ( ic . proceed ()). thenReturn ( null ) ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . invoke ( ic ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull ( ret ) ĊĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġtimes ( 2 )). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). proceed () ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" processing Ġincoming Ġmessage ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġinvoke _ message _ happy () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġmock < Incoming Kafka Record Metadata < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġheaders Ġ} Ġdo Return Ġ( Record Headers (). add ( REQUEST _ ID , Ġ" 1 2 3 ". toByteArray ()). add ( SESSION _ ID , Ġ" 2 3 4 ". toByteArray ())) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= ĠMessage . of ("{} "). add Metadata ( metadata ) ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( msg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Message Consumer Send To D lt :: class . java . getMethod (" process ", ĠMessage :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Message Consumer Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġwhenever ( ic . proceed ()). thenReturn ( completed Future ( null )) ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . invoke ( ic ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġtimes ( 2 )). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). proceed () ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" processing Ġincoming Ġmessage ") ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" kafka Ġack ed Ġmessage ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġinvoke _ message _ exception In Pro ceed () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġmock < Incoming Kafka Record Metadata < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġheaders Ġ} Ġdo Return Ġ( Record Headers (). add ( REQUEST _ ID , Ġ" 1 ". toByteArray ()). add ( SESSION _ ID , Ġ" 2 ". toByteArray ())) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= ĠMock Message ( metadata ) ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( msg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Message Consumer Send To D lt :: class . java . getMethod (" process ", ĠMessage :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Message Consumer Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġe Ġ= ĠRuntimeException (" test ") ĊĠĠĠĠĠĠĠ Ġwhenever ( ic . proceed ()). then Throw ( e ) Ġ// Ġ< --- ĠSTE ERS ĠTEST ĊĠĠĠĠĠĠĠ Ġwhenever ( error Handler . d lt ( msg , Ġe )). thenReturn ( completed Future ( null )) ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . invoke ( ic ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġmsg . ack Count ) ĊĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġtimes ( 3 )). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). method ĊĠĠĠĠĠĠĠ Ġverify ( ic ). proceed () ĊĠĠĠĠĠĠĠ Ġverify ( log ). warn (""" EH 0 0 4 Ġfailed Ġto Ġprocess Ġmessage Ġ{} Ġ- Ġsending Ġit Ġto Ġthe ĠD LT Ġwith ĠsessionId Ġ 2 """, Ġe ) ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" processing Ġincoming Ġmessage ") ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" kafka Ġack ed Ġmessage ") ĊĠĠĠĠĠĠĠ Ġverify ( error Handler ). d lt ( msg , Ġe ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġinvoke _ message _ exception In Pro ceed Result () Ġ{ Ġ// Ġie Ġin Ġthe Ġcompletion Ġstage Ġreturned Ġfrom Ġdownstream ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġmock < Incoming Kafka Record Metadata < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġheaders Ġ} Ġdo Return Ġ( Record Headers (). add ( REQUEST _ ID , Ġ" 1 ". toByteArray ()). add ( SESSION _ ID , Ġ" 2 ". toByteArray ())) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= ĠMock Message ( metadata ) ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( msg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Message Consumer Send To D lt :: class . java . getMethod (" process ", ĠMessage :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Message Consumer Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġe Ġ= ĠRuntimeException (" test ") ĊĠĠĠĠĠĠĠ Ġwhenever ( ic . proceed ()). thenReturn ( failed Future < Unit >( e )) Ġ// Ġ< --- ĠSTE ERS ĠTEST ĊĠĠĠĠĠĠĠ Ġwhenever ( error Handler . d lt ( msg , Ġe )). thenReturn ( completed Future ( null )) ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . invoke ( ic ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġmsg . ack Count ) ĊĊĠĠĠĠĠĠĠ Ġverify ( ic , Ġtimes ( 3 )). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). method ĊĠĠĠĠĠĠĠ Ġverify ( ic ). proceed () ĊĠĠĠĠĠĠĠ Ġverify ( log ). warn (""" EH 0 0 4 Ġfailed Ġto Ġprocess Ġmessage Ġ{} Ġ- Ġsending Ġit Ġto Ġthe ĠD LT Ġwith ĠsessionId Ġ 2 """, Ġe ) ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" processing Ġincoming Ġmessage ") ĊĠĠĠĠĠĠĠ Ġverify ( log ). debug (" kafka Ġack ed Ġmessage ") ĊĠĠĠĠĠĠĠ Ġverify ( error Handler ). d lt ( msg , Ġe ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġdeal With Exception If Ne cessary _ no Exception () Ġ{ ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{} ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . deal With Exception If Ne cessary ( null , Ġic , ĠContext ()) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġdeal With Exception If Ne cessary _ d lt Fails () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġmock < Incoming Kafka Record Metadata < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġheaders Ġ} Ġdo Return Ġ( Record Headers (). add ( REQUEST _ ID , Ġ" 1 ". toByteArray ()). add ( SESSION _ ID , Ġ" 2 ". toByteArray ())) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= ĠMock Message ( metadata ) ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( msg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Message Consumer Send To D lt :: class . java . getMethod (" process ", ĠMessage :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Message Consumer Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġoriginal Exception Ġ= ĠRuntimeException (" original ") ĊĠĠĠĠĠĠĠ Ġval Ġd lt Exception Ġ= ĠRuntimeException (" test ") ĊĠĠĠĠĠĠĠ Ġwhenever ( error Handler . d lt ( msg , Ġoriginal Exception )). then Throw ( d lt Exception ) Ġ// Ġ< --- ĠSTE ERS ĠTEST ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . deal With Exception If Ne cessary ( original Exception , Ġic , ĠContext . of ( RequestId (" 1 "), ĠSession Id (" 2 "), Ġmsg )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġmsg . ack Count ) Ġ// Ġack Ġis Ġcalled Ġoutside Ġof Ġmethod Ġbeing Ġtested ĊĊĠĠĠĠĠĠĠ Ġverify ( ic ). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). method ĊĠĠĠĠĠĠĠ Ġverify ( log ). warn (""" EH 0 0 4 Ġfailed Ġto Ġprocess Ġmessage Ġ{} Ġ- Ġsending Ġit Ġto Ġthe ĠD LT Ġwith ĠsessionId Ġ 2 """, Ġoriginal Exception ) ĊĠĠĠĠĠĠĠ Ġverify ( log ). error (" EH 0 0 6 a Ġfailed Ġto Ġprocess Ġmessage Ġ{} Ġ- Ġthis Ġmessage Ġis Ġbeing Ġdumped Ġhere ", Ġd lt Exception ) ĊĠĠĠĠĠĠĠ Ġverify ( log ). error (" EH 0 0 6 b Ġoriginal Ġexception Ġwas ", Ġoriginal Exception ) ĊĠĠĠĠĠĠĠ Ġverify ( error Handler ). d lt ( msg , Ġoriginal Exception ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġdeal With Exception If Ne cessary _ not To D lt () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġmock < Incoming Kafka Record Metadata < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġheaders Ġ} Ġdo Return Ġ( Record Headers (). add ( REQUEST _ ID , Ġ" 1 ". toByteArray ()). add ( SESSION _ ID , Ġ" 2 ". toByteArray ())) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= ĠMock Message ( metadata ) ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( msg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Message Consumer Dont Send To D lt :: class . java . getMethod (" process ", ĠMessage :: class . java )) Ġ// Ġ< --- ĠSTE ERS ĠTEST ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Message Consumer Dont Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġoriginal Exception Ġ= ĠRuntimeException (" original ") ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . deal With Exception If Ne cessary ( original Exception , Ġic , ĠContext . of ( RequestId (" 1 "), ĠSession Id (" 2 "), Ġmsg )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġmsg . ack Count ) Ġ// Ġack Ġis Ġcalled Ġoutside Ġof Ġmethod Ġbeing Ġtested ĊĊĠĠĠĠĠĠĠ Ġverify ( ic ). parameters ĊĠĠĠĠĠĠĠ Ġverify ( ic ). method ĊĠĠĠĠĠĠĠ Ġverify ( log ). error (""" EH 0 0 5 Ġfailed Ġto Ġprocess Ġmessage Ġ{} Ġ- ĠNOT Ġsending Ġit Ġto Ġthe ĠD LT Ġbecause Ġ@ P imp ed With D lt Ġis Ġconfigured Ġwith Ġ' send To D lt Ġ= Ġfalse ' Ġ- Ġthis Ġmessage Ġis Ġbeing Ġdumped Ġhere """, Ġoriginal Exception ) ĊĠĠĠĠĠĠĠ Ġverify ( error Handler , Ġnever ()). d lt ( any (), Ġany ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġdeal With Exception If Ne cessary _ missing SessionId () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmetadata Ġ= Ġmock < Incoming Kafka Record Metadata < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġheaders Ġ} Ġdo Return Ġ( Record Headers ()) Ġ// Ġ< --- ĠSTE ERS ĠTEST ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= ĠMock Message ( metadata ) ĊĠĠĠĠĠĠĠ Ġic Ġ= Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġparameters Ġ} Ġdo Return ( array Of ( msg )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġmethod Ġ} Ġdo Return ( Test Message Consumer Dont Send To D lt :: class . java . getMethod (" process ", ĠMessage :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġtarget Ġ} Ġdo Return ( Test Message Consumer Dont Send To D lt ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġoriginal Exception Ġ= ĠRuntimeException (" original ") ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠP imp ed And With D lt And Ack Interceptor ( error Handler , ĠContext (), ĠObjectMapper (), Ġtracer ) ĊĠĠĠĠĠĠĠ Ġsut . log Ġ= Ġlog ĊĊĠĠĠĠĠĠĠ Ġ// Ġwhen ĊĠĠĠĠĠĠĠ Ġval Ġret Ġ= Ġsut . deal With Exception If Ne cessary ( original Exception , Ġic , ĠContext . of ( RequestId (" 1 "), ĠSession Id (""), Ġmsg )) Ġ// Ġ< --- ĠSTE ERS ĠTEST ĊĊĠĠĠĠĠĠĠ Ġ// Ġthen ĊĠĠĠĠĠĠĠ ĠassertNull (( ret Ġas ĠCompletableFuture < *> ). get ()) ĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġmsg . ack Count ) Ġ// Ġack Ġis Ġcalled Ġoutside Ġof Ġmethod Ġbeing Ġtested ĊĊĠĠĠĠĠĠĠ Ġverify ( ic ). parameters ĊĠĠĠĠĠĠĠ Ġverify ( log ). error (""" EH 0 0 3 Ġfailed Ġto Ġprocess Ġmessage Ġ{} Ġ- Ġunknown ĠsessionId Ġso Ġnot Ġsending Ġit Ġto Ġthe ĠD LT Ġ- Ġthis Ġmessage Ġis Ġbeing Ġdumped Ġhere Ġ- Ġthis Ġis Ġan Ġerror Ġin Ġthe Ġprogram Ġ- Ġevery Ġmessage ĠMUST Ġhave Ġa ĠsessionId Ġheader Ġor Ġattribute Ġat Ġthe Ġroot """, Ġoriginal Exception ) ĊĠĠĠĠĠĠĠ Ġverify ( error Handler , Ġnever ()). d lt ( any (), Ġany ()) ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠTest String Consumer Send To D lt Ġ{ ĊĠĠĠ Ġ@ SuppressWarnings (" unused ") ĊĠĠĠ Ġ@ P imp ed And With D lt And Ack ĊĠĠĠ Ġfun Ġprocess ( s : ĠString ) Ġ= ĠUnit Ċ } Ċ Ċ class ĠTest String Consumer Dont Send To D lt Ġ{ ĊĠĠĠ Ġ@ SuppressWarnings (" unused ") ĊĠĠĠ Ġ@ P imp ed And With D lt And Ack ( false ) ĊĠĠĠ Ġfun Ġprocess ( s : ĠString ) Ġ= ĠUnit Ċ } Ċ Ċ class ĠTest Message Consumer Send To D lt Ġ{ ĊĠĠĠ Ġ@ SuppressWarnings (" unused ") ĊĠĠĠ Ġ@ P imp ed And With D lt And Ack ĊĠĠĠ Ġfun Ġprocess ( m : ĠMessage < String >) Ġ= ĠUnit Ċ } Ċ Ċ class ĠTest Message Consumer Dont Send To D lt Ġ{ ĊĠĠĠ Ġ@ SuppressWarnings (" unused ") ĊĠĠĠ Ġ@ P imp ed And With D lt And Ack ( false ) ĊĠĠĠ Ġfun Ġprocess ( m : ĠMessage < String >) Ġ= ĠUnit Ċ } Ċ Ċ class ĠTest Consumer Wrong Num Params Ġ{ ĊĠĠĠ Ġ@ P imp ed And With D lt And Ack ( false ) ĊĠĠĠ Ġfun Ġprocess () Ġ= ĠUnit Ċ } Ċ Ċ class ĠTest Consumer Wrong Param Type Ġ{ ĊĠĠĠ Ġ@ SuppressWarnings (" unused ") ĊĠĠĠ Ġ@ P imp ed And With D lt And Ack ( false ) ĊĠĠĠ Ġfun Ġprocess ( i : ĠInt ) Ġ= ĠUnit Ċ } Ċ Ċ class ĠMock Message ( k Metadata : ĠIncoming Kafka Record Metadata < String , ĠString >) Ġ: ĠMessage < String > Ġ{ ĊĠĠĠ Ġvar Ġack Count Ġ= Ġ 0 ĊĠĠĠ Ġval Ġmd : ĠMetadata ĊĊĠĠĠ Ġinit { ĊĠĠĠĠĠĠĠ Ġmd Ġ= Ġsuper . getMetadata (). with ( k Metadata ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Payload (): ĠString Ġ= Ġ"{} " ĊĊĠĠĠ Ġoverride Ġfun Ġack (): ĠCompletion Stage < Void > Ġ{ ĊĠĠĠĠĠĠĠ Ġack Count ++ ĊĠĠĠĠĠĠĠ Ġreturn Ġcompleted Future ( null ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Metadata (): ĠMetadata Ġ= Ġmd Ċ }