/* Ċ Ġ* ĠCopyright Ġ 2 0 1 8 ĠThe ĠHy ve Ċ Ġ* Ċ Ġ* Ġ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 . Ċ Ġ*/ Ċ Ċ package Ġorg . radar base . output . path Ċ Ċ import Ġorg . apache . avro . generic . Generic Record Ċ import Ġorg . radar base . output . Plugin Ċ import Ġorg . radar base . output . util . Time Util Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . nio . file . Path Ċ import Ġjava . time . Instant Ċ import Ġjava . time . Zone Offset . UTC Ċ import Ġjava . time . format . DateTime Formatter Ċ import Ġjava . util . regex . Pattern Ċ Ċ abstract Ġclass ĠRecord Path Factory Ġ: ĠPlugin Ġ{ ĊĠĠĠ Ġlateinit Ġvar Ġroot : ĠPath ĊĠĠĠ Ġlateinit Ġvar Ġextension : ĠString ĊĊĠĠĠ Ġprotected Ġopen Ġvar Ġtime Bin Format : ĠDateTime Formatter Ġ= ĠHO URL Y _ TIME _ BIN _ FORMAT ĊĊĠĠĠ Ġoverride Ġfun Ġinit ( properties : ĠMap < String , ĠString >) Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . init ( properties ) ĊĠĠĠĠĠĠĠ Ġproperties [" time Bin Format "] ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtime Bin Format Ġ= ĠDateTime Formatter ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. of Pattern ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Zone ( UTC ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠIllegalArgumentException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Cannot Ġuse Ġtime Ġbin Ġformat Ġ{}, Ġusing Ġ{} Ġinstead ", Ġit , Ġtime Bin Format , Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠGet Ġthe Ġorganization Ġof Ġgiven Ġrecord Ġin Ġgiven Ġtopic . ĊĠĠĠĠ Ġ* Ġ@ param Ġtopic ĠKafka Ġtopic Ġname ĊĠĠĠĠ Ġ* Ġ@ param Ġrecord Ġrecord Ġwith Ġpossible Ġkey Ġand Ġvalue Ġfields Ġcontaining Ġrecords ĊĠĠĠĠ Ġ* Ġ@ param Ġattempt Ġnumber Ġof Ġprevious Ġattempts Ġto Ġwrite Ġgiven Ġrecord . ĠThis Ġincreases Ġif Ġprevious ĊĠĠĠĠ Ġ* Ġpaths Ġalready Ġexisted Ġand Ġare Ġincompatible . ĊĠĠĠĠ Ġ* Ġ@ return Ġorganization Ġof Ġgiven Ġrecord ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġopen Ġfun Ġget Record Organization ( ĊĠĠĠĠĠĠĠ Ġtopic : ĠString , ĊĠĠĠĠĠĠĠ Ġrecord : ĠGeneric Record , ĊĠĠĠĠĠĠĠ Ġattempt : ĠInt , ĊĠĠĠ Ġ): ĠRecord Organization Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġkey Field Ġ= Ġrecord . get (" key ") Ġas ? ĠGeneric Record ĊĠĠĠĠĠĠĠ Ġval Ġvalue Field Ġ= Ġrecord . get (" value ") Ġas ? ĠGeneric Record ĊĊĠĠĠĠĠĠĠ Ġif Ġ( key Field Ġ== Ġnull Ġ|| Ġvalue Field Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Failed Ġto Ġprocess Ġ{}", Ġrecord ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalArgumentException (" Failed Ġto Ġprocess Ġ$ record ; Ġno Ġkey Ġor Ġvalue ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġtime Ġ= ĠTime Util . getDate ( key Field , Ġvalue Field ) ĊĊĠĠĠĠĠĠĠ Ġval Ġrelative Path Ġ= Ġget RelativePath ( topic , Ġkey Field , Ġvalue Field , Ġtime , Ġattempt ) ĊĠĠĠĠĠĠĠ Ġval Ġoutput Path Ġ= Ġroot . resolve ( relative Path ) ĊĠĠĠĠĠĠĠ Ġval Ġcategory Ġ= Ġget Category ( key Field , Ġvalue Field ) ĊĠĠĠĠĠĠĠ Ġreturn ĠRecord Organization ( output Path , Ġcategory , Ġtime ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠGet Ġthe Ġrelative Ġpath Ġcorresponding Ġto Ġgiven Ġrecord Ġon Ġgiven Ġtopic . ĊĠĠĠĠ Ġ* Ġ@ param Ġtopic ĠKafka Ġtopic Ġname ĊĠĠĠĠ Ġ* Ġ@ param Ġkey Ġrecord Ġkey ĊĠĠĠĠ Ġ* Ġ@ param Ġvalue Ġrecord Ġvalue ĊĠĠĠĠ Ġ* Ġ@ param Ġtime Ġtime Ġcontained Ġin Ġthe Ġrecord ĊĠĠĠĠ Ġ* Ġ@ param Ġattempt Ġnumber Ġof Ġprevious Ġattempts Ġto Ġwrite Ġgiven Ġrecord . ĠThis Ġincreases Ġif Ġprevious ĊĠĠĠĠ Ġ* Ġpaths Ġalready Ġexisted Ġand Ġare Ġincompatible . ĊĠĠĠĠ Ġ* Ġ@ return Ġrelative Ġpath Ġcorresponding Ġto Ġgiven Ġparameters . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġabstract Ġfun Ġget RelativePath ( ĊĠĠĠĠĠĠĠ Ġtopic : ĠString , ĊĠĠĠĠĠĠĠ Ġkey : ĠGeneric Record , ĊĠĠĠĠĠĠĠ Ġvalue : ĠGeneric Record , ĊĠĠĠĠĠĠĠ Ġtime : ĠInstant ?, ĊĠĠĠĠĠĠĠ Ġattempt : ĠInt , ĊĠĠĠ Ġ): ĠPath ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠGet Ġthe Ġcategory Ġof Ġa Ġrecord , Ġrepresenting Ġa Ġpartition ing Ġfor Ġa Ġgiven Ġtopic Ġand Ġuser . ĊĠĠĠĠ Ġ* Ġ@ param Ġkey Ġrecord Ġkey ĊĠĠĠĠ Ġ* Ġ@ param Ġvalue Ġrecord Ġvalue ĊĠĠĠĠ Ġ* Ġ@ return Ġcategory Ġname . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġabstract Ġfun Ġget Category ( key : ĠGeneric Record , Ġvalue : ĠGeneric Record ): ĠString ĊĊĠĠĠ Ġopen Ġfun ĠgetTime Bin ( time : ĠInstant ?): ĠString Ġ= Ġtime ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ? . let Ġ{ Ġtime Bin Format . format ( time )} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġ" unknown _ date " ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠOrganization Ġof Ġa Ġrecord . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġdata Ġclass ĠRecord Organization ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ/** ĠPath Ġthat Ġthe Ġrecord Ġshould Ġbe Ġstored Ġin . Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpath : ĠPath , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ/** ĠCategory Ġor Ġpartition Ġthat Ġthe Ġrecord Ġbelongs Ġto . Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcategory : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ/** ĠTime Ġcontained Ġin Ġthe Ġrecord , Ġif Ġany . Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtime : ĠInstant ?) ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Record Path Factory :: class . java ) ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠIL LEG AL _ CHARACTER _ PATTERN Ġ= ĠPattern . compile ("[ ^ a - zA - Z 0 - 9 _- ]+ ") ĊĊĠĠĠĠĠĠĠ Ġval ĠHO URL Y _ TIME _ BIN _ FORMAT : ĠDateTime Formatter Ġ= ĠDateTime Formatter ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. of Pattern (" yyyy MM dd _ HH ' 0 0 '") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Zone ( UTC ) ĊĊĠĠĠĠĠĠĠ Ġfun Ġsanitize Id ( id : ĠAny ?, ĠdefaultValue : ĠString ): ĠString Ġ= Ġid ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ? . let Ġ{ ĠIL LEG AL _ CHARACTER _ PATTERN . matcher ( it . toString ()). replaceAll ("") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ? . take If Ġ{ Ġit . isNotEmpty () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: ĠdefaultValue ĊĠĠĠ Ġ} Ċ } Ċ