package Ġapp . load . services Ċ import Ġapp . load . helpers . Secret Helper Interface Ċ import Ġapp . load . configurations . Metadata Store Configuration Ċ import Ġapp . load . domain . H Base Payload Ċ import Ġapp . load . helpers . impl . AW SS ec ret Helper Ċ import Ġapp . load . helpers . impl . Dummy Secret Helper Ċ import Ġapp . load . utility . TextUtils Ċ import Ġorg . apache . commons . db cp 2 . Basic DataSource Ċ import Ġuk . gov . d wp . data works . logging . Data works Logger Ċ import Ġjava . sql . Connection Ċ import Ġjava . util .* Ċ import Ġkotlin . system . measure Time Millis Ċ import Ġkotlin . time . Experimental Time Ċ import Ġkotlin . time . hours Ċ Ċ @ Experimental Time Ċ class ĠMetadata Store Service ( private Ġval Ġconnection : ĠConnection ): ĠAuto Closeable Ġ{ ĊĊĠĠĠ Ġfun Ġrecord Batch ( payload s : ĠList < H Base Payload >) Ġ{ ĊĠĠĠĠĠĠĠ Ġlogger . info (" Put ting Ġbatch Ġinto Ġmetadata Ġstore , Ġsize Ġ${ payload s . size }") ĊĠĠĠĠĠĠĠ Ġval Ġtime Taken Ġ= Ġmeasure Time Millis Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith ( record Processing Attempt Statement ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpayload s . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset String ( 1 , Ġtext Utils . print able Key ( it . key )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Long ( 2 , Ġit . version ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset String ( 3 , Ġit . topic ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Int ( 4 , Ġit . partition ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Long ( 5 , Ġit . offset ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd Batch () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexecute Batch () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġprintln (" Put Ġbatch Ġinto Ġmetadata Ġstore , Ġtime _ taken Ġ$ time Taken , Ġsize : Ġ${ payload s . size }") ĊĠĠĠĠĠĠĠ Ġlogger . info (" Put Ġbatch Ġinto Ġmetadata Ġstore Ġtime _ taken Ġ$ time Taken , Ġsize Ġ${ payload s . size }") ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġprivate Ġval Ġrecord Processing Attempt Statement Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġconnection . prepareStatement (""" ĊĠĠĠĠĠĠĠĠĠĠĠ ĠINSERT ĠINTO Ġ${ Metadata Store Configuration . metadata Store Table } Ġ( hbase _ id , Ġh base _ timestamp , Ġtopic _ name , Ġkafka _ partition , Ġkafka _ offset ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠVALUES Ġ( ?, Ġ?, Ġ?, Ġ?, Ġ? ) ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġis Using AWS Ġ= ĠMetadata Store Configuration . use Aws Secrets ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsecret Helper : ĠSecret Helper Interface Ġ= Ġ Ġif Ġ( is Using AWS ) ĠAW SS ec ret Helper () Ġelse ĠDummy Secret Helper () ĊĊĠĠĠĠĠĠĠ Ġfun Ġconnect (): ĠMetadata Store Service Ġ= ĠMetadata Store Service ( dataSource . connection ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠdataSource Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( database Url , Ġdatabase Properties ) Ġ= Ġconnection Properties () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠBasic DataSource (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġurl Ġ= Ġdatabase Url ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalidation Query Ġ= Ġ" SELECT Ġcount (*) Ġfrom Ġ${ Metadata Store Configuration . metadata Store Table }" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmax Conn Lifetime Millis Ġ= Ġ 1 2 . hours . to Long Milliseconds () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdatabase Properties . forEach Ġ{ Ġ( name , Ġvalue ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd Connection Property ( name . toString (), Ġvalue . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġconnection Properties (): ĠPair < String , ĠProperties > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhostname Ġ= ĠMetadata Store Configuration . properties [" rds . endpoint "] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġport Ġ= ĠMetadata Store Configuration . properties [" rds . port "] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjdbc Url Ġ= Ġ" jdbc : mysql :// $ hostname :$ port /${ Metadata Store Configuration . properties . getProperty (" database ") }" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproperties With Password : ĠProperties Ġ= ĠMetadata Store Configuration . properties . clone () Ġas ĠProperties ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproperties With Password [" password "] Ġ= Ġpassword ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠPair ( jdbc Url , Ġproperties With Password ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġpassword Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsecret Name Ġ= ĠMetadata Store Configuration . properties . getProperty (" rds . password . secret . name ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsecret Helper . get Secret ( secret Name ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġlogger Ġ= ĠData works Logger . getLogger ( Metadata Store Service :: class . toString ()) ĊĠĠĠĠĠĠĠ Ġval Ġtext Utils Ġ= ĠText Utils () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ= Ġconnection . close () Ċ } Ċ