package Ġmo ip . mocked k Ċ Ċ import Ġorg . apache . kafka . common . serialization . Deserializer Ċ import Ġorg . apache . kafka . common . serialization . Serializer Ċ import Ġorg . apache . kafka . streams . Streams Config Ċ import Ġorg . apache . kafka . streams . processor . Topology Builder Ċ import Ġorg . apache . kafka . test . Processor Topology Test Driver Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util .* Ċ Ċ /** Ċ Ġ* ĠCreated Ġby Ġdie go ic osta Ġon Ġ 1 2 / 1 0 / 1 6 . Ċ Ġ*/ Ċ class ĠMock ed Kafka Ġ{ ĊĊĠĠĠ Ġfun Ġapply ( topology : ĠTop ology Builder ): ĠInputStream Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġextractor Ġ= ĠMeta Topology ( topology ) ĊĠĠĠĠĠĠĠ Ġreturn ĠInputStream ( extractor ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠInputStream ( private Ġval Ġtopology Extractor : ĠMeta Topology ) Ġ{ ĊĊĠĠĠ Ġfun Ġ< K Ġ: ĠAny , ĠV Ġ: ĠAny > Ġinput ( key : ĠK , Ġvalue : ĠV ): ĠSource Serializer < K , ĠV > Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġinputs Ġ= ĠData Stream < K , ĠV >(). add ( key , Ġvalue ) ĊĊĠĠĠĠĠĠĠ Ġreturn ĠSource Serializer ( inputs , Ġtopology Extractor ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠSource Serializer < K Ġ: ĠAny , ĠV Ġ: ĠAny >( private Ġval Ġinputs : ĠData Stream < K , ĠV >, Ġprivate Ġval Ġtopology : ĠMeta Topology ) Ġ{ ĊĊĠĠĠ Ġfun Ġserialized By ( key Serializer : ĠSerializer < K >, Ġvalue Serializer : ĠSerializer < V >): ĠOutputStream < K , ĠV > Ġ{ ĊĠĠĠĠĠĠĠ Ġinputs . serialize ( key Serializer , Ġvalue Serializer ) ĊĠĠĠĠĠĠĠ Ġreturn ĠOutputStream ( topology , Ġinputs ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġinput ( key : ĠK , Ġvalue : ĠV ): ĠSource Serializer < K , ĠV > Ġ{ ĊĠĠĠĠĠĠĠ Ġinputs . add ( key , Ġvalue ) ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠOutputStream < K Ġ: ĠAny , ĠV Ġ: ĠAny >( private Ġval Ġtopology Extractor : ĠMeta Topology , Ġprivate Ġval Ġinput : ĠData Stream < K , ĠV >) Ġ{ ĊĊĠĠĠ Ġval ĠLOG Ġ= ĠLoggerFactory . getLogger ( OutputStream :: class . java ) ĊĊĠĠĠ Ġfun Ġ< K 2 Ġ: ĠAny , ĠV 2 Ġ: ĠAny > Ġoutput ( key : ĠK 2 , Ġvalue : ĠV 2 ): ĠS ink Serializer < K , ĠV , ĠK 2 , ĠV 2 > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġoutput Ġ= ĠData Stream < K 2 , ĠV 2 >(). add ( key , Ġvalue ) ĊĠĠĠĠĠĠĠ Ġreturn ĠS ink Serializer ( input , Ġoutput , Ġtopology Extractor ) ĊĠĠĠ Ġ} Ċ Ċ } Ċ Ċ class ĠData Stream < K , ĠV > Ġ{ ĊĊĠĠĠ Ġdata Ġclass ĠKey Value < K , ĠV >( val Ġkey : ĠK , Ġval Ġvalue : ĠV ) ĊĊĠĠĠ Ġprivate Ġval Ġlist : ĠMutable List < KeyValue < K , ĠV >> Ġ= Ġmutable ListOf () ĊĠĠĠ Ġprivate Ġval Ġbyte List : ĠMutable List < KeyValue < ByteArray , ĠByteArray >> Ġ= Ġmutable ListOf () ĊĊĠĠĠ Ġfun Ġadd ( key : ĠK , Ġvalue : ĠV ): ĠData Stream < K , ĠV > Ġ{ ĊĠĠĠĠĠĠĠ Ġlist . add ( KeyValue ( key , Ġvalue )) ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġge Serialized List (): ĠList < KeyValue < ByteArray , ĠByteArray >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġbyte List . toList () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġge Deserial ized List (): ĠList < KeyValue < K , ĠV >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġlist . toList () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġserialize ( key Serializer : ĠSerializer < K >, Ġvalue Serializer : ĠSerializer < V >) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor Ġ( item Ġin Ġthis . list ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbyte List . add ( KeyValue ( key Serializer . serialize ("", Ġitem . key ), Ġvalue Serializer . serialize ("", Ġitem . value ))) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ Ċ class ĠEmpty Sink Error ( msg : ĠString ) Ġ: ĠRuntimeException ( msg ) Ċ class ĠNo Match Result ( msg : ĠString ) Ġ: ĠRuntimeException ( msg ) Ċ Ċ class ĠS ink Serializer < K 1 , ĠV 1 , ĠK 2 , ĠV 2 >( private Ġval Ġinput : ĠData Stream < K 1 , ĠV 1 >, Ġprivate Ġval Ġoutput : ĠData Stream < K 2 , ĠV 2 >, Ġprivate Ġval Ġtopology : ĠMeta Topology ) Ġ{ ĊĊĠĠĠ Ġfun Ġdeserialized By ( key Deserializer : ĠDeserial izer < K 2 >, Ġvalue Deserializer : ĠDeserial izer < V 2 >) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġprops Ġ= ĠProperties () ĊĠĠĠĠĠĠĠ Ġprops . put ( Streams Config . APPLICATION _ ID _ CONFIG , Ġ" mocked -${ UUID . randomUUID () }") ĊĠĠĠĠĠĠĠ Ġprops . put ( Streams Config . BOOT STR AP _ SERVER S _ CONFIG , Ġ" localhost : 9 0 9 2 ") ĊĠĠĠĠĠĠĠ Ġval Ġdriver Ġ= ĠProcessor Topology Test Driver ( Streams Config ( props ), Ġtopology . topology ) ĊĊĠĠĠĠĠĠĠ Ġfor Ġ( i Ġin Ġinput . ge Serialized List ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdriver . process ( topology . source Topic , Ġi . key , Ġi . value ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfor Ġ( out Ġin Ġoutput . ge Deserial ized List ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġdriver . read Output ( topology . sink Topic ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord Ġ?: Ġthrow ĠEmpty Sink Error (" topic Ġ'${ topology . sink Topic }' Ġis Ġempty ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrec Out Key Ġ= Ġkey Deserializer . deserialize ("", Ġrecord . key ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrec Out Val Ġ= Ġvalue Deserializer . deserialize ("", Ġrecord . value ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Ġrec Out Key Ġ!= Ġout . key Ġ|| Ġrec Out Val Ġ!= Ġout . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠNo Match Result (" topic Ġ'${ topology . sink Topic }' Ġhas Ġa Ġdifferent Ġkey / value ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġoutput ( key : ĠK 2 , Ġvalue : ĠV 2 ): ĠS ink Serializer < K 1 , ĠV 1 , ĠK 2 , ĠV 2 > Ġ{ ĊĠĠĠĠĠĠĠ Ġoutput . add ( key , Ġvalue ) ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} Ċ Ċ } Ċ Ċ data Ġclass ĠMeta Topology ( val Ġtopology : ĠTop ology Builder ) Ġ{ ĊĠĠĠ Ġval Ġsource Topic Ġ= Ġtopology . build ( null ). source Topics (). last () ĊĠĠĠ Ġ// ĠĠĠ Ġval Ġsource Topic : ĠString Ġ= Ġtopology . source Topics (). last () Ġ Ġolder Ġk - stream Ġversion ĊĠĠĠ Ġval Ġsink Topic : ĠString Ġ= Ġtopology . build ( null ). sink Topics (). last () Ċ } ĊĊ