package Ġno . nav . hel se . h ops . test Ċ Ċ import Ġno . nav . hel se . h ops . plugin . Kafka Config Ċ import Ġno . nav . hel se . h ops . plugin . Kafka Factory Ċ import Ġorg . apache . kafka . clients . producer . Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . common . header . internals . Record Header Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util . UUID Ċ Ċ private Ġval Ġlog Ġ= ĠLoggerFactory . getLogger (" test ") Ċ private Ġval Ġheader Ġ= ĠRecord Header (" Content - Type ", Ġ" application / fhir + json ; Ġfhir Version = 4 . 0 ". toByteArray ()) Ċ Ċ class ĠEmbedded Kafka ( topic : ĠString ) Ġ: ĠAuto Closeable Ġ{ ĊĠĠĠ Ġprivate Ġval Ġkafka Ġ= ĠThread Safe Kafka Environment ( topic Names Ġ= ĠlistOf ( topic ), Ġauto Start Ġ= Ġtrue ) ĊĠĠĠ Ġprivate Ġval Ġproducer Ġ= Ġkafka . create Producer () ĊĠĠĠ Ġval Ġbro kers URL Ġget () Ġ= Ġkafka . bro kers URL ĊĊĠĠĠ Ġfun Ġproduce ( topic : ĠString , Ġkey : ĠUUID , Ġvalue : ĠByteArray ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= ĠProducer Record ( topic , Ġnull , Ġkey , Ġvalue , ĠlistOf ( header )) ĊĠĠĠĠĠĠĠ Ġproducer . send ( record ). get (). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Pro duced Ġrecord Ġon Ġ$ it Ġ( topic - partition @ offset )") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info ( record . toString ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġproducer . close () ĊĠĠĠĠĠĠĠ Ġkafka . close () ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġfun ĠThread Safe Kafka Environment . create Producer (): ĠProducer < UUID , ĠByteArray > Ġ{ ĊĠĠĠ Ġval Ġconfig Ġ= ĠKafka Config ( ĊĠĠĠĠĠĠĠ Ġbro kers Ġ= Ġbro kers URL , ĊĠĠĠĠĠĠĠ ĠgroupId Ġ= Ġ"", ĊĠĠĠĠĠĠĠ Ġtopic Ġ= Ġ"", ĊĠĠĠĠĠĠĠ ĠclientId Ġ= Ġ" producer - test - client ", ĊĠĠĠĠĠĠĠ Ġsecurity Ġ= Ġfalse , ĊĠĠĠĠĠĠĠ Ġtrust store Path Ġ= Ġ"", ĊĠĠĠĠĠĠĠ Ġkeystore Path Ġ= Ġ"", ĊĠĠĠĠĠĠĠ Ġcred store P sw Ġ= Ġ"" ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġreturn ĠKafka Factory . create Fhir Producer ( config ) Ċ } Ċ