package Ġstreams . kafka . connect . source Ċ Ċ import Ġorg . apache . kafka . connect . data . Schema Ċ import Ġorg . apache . kafka . connect . source . Source Record Ċ import Ġorg . neo 4 j . driver . Record Ċ import Ġstreams . kafka . connect . utils . as Json String Ċ import Ġstreams . kafka . connect . utils . as Struct Ċ import Ġkotlin . properties . Deleg ates Ċ Ċ class ĠSource Record Builder Ġ{ ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġtopic : ĠString ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġstreaming Property : ĠString ĊĊĠĠĠ Ġprivate Ġvar Ġtimestamp Ġby ĠDeleg ates . notNull < Long >() ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġsource Partition : ĠMap < String , ĠAny > ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġrecord : ĠRecord ĊĊĠĠĠ Ġprivate Ġvar Ġenforce Schema : ĠBoolean Ġ= Ġfalse ĊĊĠĠĠ Ġfun Ġwith Topic ( topic : ĠString ): ĠSource Record Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . topic Ġ= Ġtopic ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwith Streaming Property ( streaming Property : ĠString ): ĠSource Record Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . streaming Property Ġ= Ġstreaming Property ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwith Timestamp ( timestamp : ĠLong ): ĠSource Record Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . timestamp Ġ= Ġtimestamp ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwith Source Partition ( source Partition : ĠMap < String , ĠAny >): ĠSource Record Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . source Partition Ġ= Ġsource Partition ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwith Record ( record : ĠRecord ): ĠSource Record Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . record Ġ= Ġrecord ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwith En force Schema ( enforce Schema : ĠBoolean ): ĠSource Record Builder Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . enforce Schema Ġ= Ġenforce Schema ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġbuild (): ĠSource Record Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsource Offset Ġ= Ġmap Of (" property " Ġto Ġstreaming Property . if Blank Ġ{ Ġ" undefined " Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" value " Ġto Ġtimestamp ) ĊĠĠĠĠĠĠĠ Ġval Ġ( struct , Ġschema ) Ġ= Ġwhen Ġ( enforce Schema ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġst Ġ= Ġrecord . as Struct () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsc Ġ= Ġst . schema () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġst Ġto Ġsc ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġrecord . as Json String () Ġto ĠSchema . STRING _ SCHEMA ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠSource Record ( source Partition , Ġsource Offset , Ġtopic , Ġschema , Ġstruct , Ġschema , Ġstruct ) ĊĠĠĠ Ġ} Ċ } Ċ