package Ġmicro . apps . k stream . serializer ĊĊ Ċ import Ġcom . github . avro kotlin . avro 4 k . Av ro Ċ import Ġcom . github . avro kotlin . avro 4 k . io . Av ro Encode Format Ċ import Ġcom . github . th ake . kafka . avro 4 k . serializer . Av ro 4 k Schema Utils Ċ import Ġio . con fluent . kafka . serializers . Non Record Container Ċ import Ġkotlinx . serialization . Internal Serialization Api Ċ import Ġkotlinx . serialization . K Serializer Ċ import Ġkotlinx . serialization . serializer Ċ import Ġorg . apache . avro . Schema Ċ import Ġorg . apache . avro . generic . Generic Datum Writer Ċ import Ġorg . apache . avro . io . Encoder Factory Ċ import Ġorg . apache . kafka . common . errors . Serialization Exception Ċ import Ġjava . io . ByteArray OutputStream Ċ import Ġjava . io . IOException Ċ import Ġjava . nio . ByteBuffer Ċ Ċ abstract Ġclass ĠAbstract Crypto Kafka Av ro 4 k Serializer Ġ: ĠAbstract Crypto Kafka Av ro 4 k Ser De () Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġauto Register Schema Ġ= Ġfalse ĊĊĠĠĠ Ġprotected Ġval Ġav ro Schema Utils Ġ= ĠAv ro 4 k Schema Utils () ĊĠĠĠ Ġprotected Ġfun Ġconfigure ( config : ĠCrypto Kafka Av ro 4 k Serializer Config ) Ġ{ ĊĠĠĠĠĠĠĠ Ġauto Register Schema Ġ= Ġconfig . auto Register Schema () ĊĠĠĠĠĠĠĠ Ġsuper . configure ( config ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprotected Ġfun Ġserializer Config ( props : ĠMap < String , Ġ* >): ĠCrypto Kafka Av ro 4 k Serializer Config Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠCrypto Kafka Av ro 4 k Serializer Config ( props ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( Serialization Exception :: class ) ĊĠĠĠ Ġprotected Ġfun Ġserialize Impl ( subject : ĠString ?, Ġobj : ĠAny ?): ĠByteArray ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( obj Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcurrent Schema Ġ= Ġav ro Schema Utils . getSchema ( obj ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġid Ġ= Ġget Schema Id ( subject , Ġcurrent Schema ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġout Ġ= ĠByteArrayOutputStream () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwrite Schema Id ( out , Ġid ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( obj Ġis ĠByteArray ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġout . write ( obj ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserialize Value ( out , Ġobj , Ġcurrent Schema ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbytes Ġ= Ġout . toByteArray () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġout . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbytes ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( re : ĠRuntimeException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠSerialization Exception (" Error Ġserial izing ĠAv ro Ġmessage Ġwith Ġav ro 4 k ", Ġre ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( io : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠSerialization Exception (" Error Ġserial izing ĠAv ro Ġmessage Ġwith Ġav ro 4 k ", Ġio ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Opt In ( Internal Serialization Api :: class ) ĊĠĠĠ Ġfun Ġserialize Value ( out : ĠByteArrayOutputStream , Ġobj : ĠAny , Ġcurrent Schema : ĠSchema ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġvalue Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( obj Ġis ĠNon Record Container ) Ġobj . value Ġelse Ġobj ĊĠĠĠĠĠĠĠ Ġif Ġ( current Schema . type Ġ== ĠSchema . Type . RECORD ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAv ro . default . open OutputStream ( value :: class . serializer () Ġas ĠK Serializer < Any >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġencode Format Ġ= ĠAv ro Encode Format . Binary ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschema Ġ= Ġcurrent Schema ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. to ( out ). write ( obj ). close () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġencoder Ġ= ĠEncoder Factory . get (). direct Binary Encoder ( out , Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdatum Writer Ġ= ĠGeneric Datum Writer < Any >( current Schema ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdatum Writer . write ( obj , Ġencoder ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġencoder . flush () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġwrite Schema Id ( out : ĠByteArrayOutputStream , Ġid : ĠInt ) Ġ{ ĊĠĠĠĠĠĠĠ Ġout . write ( 0 ) ĊĠĠĠĠĠĠĠ Ġout . write ( ByteBuffer . allocate ( 4 ). put Int ( id ). array ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Schema Id ( ĊĠĠĠĠĠĠĠ Ġsubject : ĠString ?, ĊĠĠĠĠĠĠĠ Ġschema : ĠSchema ? ĊĠĠĠ Ġ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( auto Register Schema ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġregister With Retry ( subject , Ġschema ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget Schema Id With Retry ( subject , Ġschema ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ