/** Ċ Ġ* ĠCopyright Ġ 2 0 1 9 ĠCon fluent ĠInc . Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ @ file : Jvm Name (" Producer Example ") Ċ Ċ package Ġio . con fluent . examples . clients . cloud Ċ Ċ import Ġio . con fluent . examples . clients . cloud . model . Data Record Ċ import Ġio . con fluent . examples . clients . cloud . util . load Config Ċ import Ġio . con fluent . kafka . serializers . Kafka Json Serializer Ċ import Ġorg . apache . kafka . clients . admin . Admin Client Ċ import Ġorg . apache . kafka . clients . admin . New Topic Ċ import Ġorg . apache . kafka . clients . producer . Kafka Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Config .* Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . clients . producer . Record Metadata Ċ import Ġorg . apache . kafka . common . errors . Topic Exists Exception Ċ import Ġorg . apache . kafka . common . serialization . String Serializer Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . ExecutionException Ċ import Ġkotlin . system . exit Process Ċ Ċ // ĠCreate Ġtopic Ġin ĠCon fluent ĠCloud Ċ fun Ġcreate Topic ( topic : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpartitions : ĠInt , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreplication : ĠShort , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcloud Config : ĠProperties ) Ġ{ ĊĠ Ġval Ġnew Topic Ġ= ĠNew Topic ( topic , Ġpartitions , Ġreplication ) ĊĊĊĠ Ġtry Ġ{ ĊĠĠĠ Ġwith ( Admin Client . create ( cloud Config )) Ġ{ ĊĠĠĠĠĠ Ġcreate Topics ( listOf ( new Topic )). all (). get () ĊĠĠĠ Ġ} ĊĠ Ġ} Ġcatch Ġ( e : ĠExecution Exception ) Ġ{ ĊĠĠĠ Ġif Ġ( e . cause Ġ! is ĠTopic Exists Exception ) Ġthrow Ġe ĊĠ Ġ} Ċ } Ċ Ċ fun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠ Ġif Ġ( args . size Ġ!= Ġ 2 ) Ġ{ ĊĠĠĠ Ġprintln (" Please Ġprovide Ġcommand Ġline Ġarguments : Ġconfig Path Ġtopic ") ĊĠĠĠ Ġexit Process ( 1 ) ĊĠ Ġ} ĊĊĠ Ġ// ĠLoad Ġproperties Ġfrom Ġfile ĊĠ Ġval Ġprops Ġ= Ġload Config ( args [ 0 ]) ĊĊĠ Ġ// ĠCreate Ġtopic Ġif Ġneeded ĊĠ Ġval Ġtopic Ġ= Ġargs [ 1 ] ĊĠ Ġcreate Topic ( topic , Ġ 1 , Ġ 3 , Ġprops ) ĊĊĠ Ġ// ĠAdd Ġadditional Ġproperties . ĊĠ Ġprops [ ACK S _ CONFIG ] Ġ= Ġ" all " ĊĠ Ġprops [ KEY _ SERIAL IZER _ CLASS _ CONFIG ] Ġ= ĠString Serializer :: class . qualified Name ĊĠ Ġprops [ VALUE _ SERIAL IZER _ CLASS _ CONFIG ] Ġ= ĠKafka Json Serializer :: class . qualified Name ĊĊĠ Ġ// ĠPro duce Ġsample Ġdata ĊĠ Ġval Ġnum Messages Ġ= Ġ 1 0 ĊĠ Ġ// Ġ` use ` Ġwill Ġexecute Ġblock Ġand Ġclose Ġproducer Ġautomatically ĠĠĊĠ ĠKafka Producer < String , ĠData Record >( props ). use Ġ{ Ġproducer Ġ-> ĊĠĠĠ Ġrepeat ( num Messages ) Ġ{ Ġi Ġ-> ĊĠĠĠĠĠ Ġval Ġkey Ġ= Ġ" alice " ĊĠĠĠĠĠ Ġval Ġrecord Ġ= ĠData Record ( i . to Long ()) ĊĠĠĠĠĠ Ġprintln (" Pro ducing Ġrecord : Ġ$ key \ t $ record ") ĊĊĠĠĠĠĠ Ġproducer . send ( Producer Record ( topic , Ġkey , Ġrecord )) Ġ{ Ġm : ĠRecord Metadata , Ġe : ĠException ? Ġ-> ĊĠĠĠĠĠĠĠ Ġwhen Ġ( e ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġno Ġexception , Ġgood Ġto Ġgo ! ĊĠĠĠĠĠĠĠĠĠ Ġnull Ġ-> Ġprintln (" Pro duced Ġrecord Ġto Ġtopic Ġ${ m . topic ()} Ġpartition Ġ[ ${ m . partition ()} ] Ġ@ Ġoffset Ġ${ m . offset () }") ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġprint Ġstack trace Ġin Ġcase Ġof Ġexception ĊĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġe . printStackTrace () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġproducer . flush () ĊĠĠĠ Ġprintln (" 1 0 Ġmessages Ġwere Ġproduced Ġto Ġtopic Ġ$ topic ") ĊĠ Ġ} Ċ Ċ } ĊĊĊ