/* Ċ Ġ* ĠCopyright Ġ 2 0 2 1 - 2 0 2 2 ĠMicro ĠFocus Ġor Ġone Ġof Ġits Ġaffiliates Ċ Ġ* Ċ Ġ* Ġ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 Ġcom . micro focus . fl ork . kubernetes . api . v 1 . re conc ilers Ċ Ċ import Ġcom . micro focus . fl ork . kubernetes . api . v 1 . model . F link Job Custom Resource Ċ import Ġcom . micro focus . fl ork . kubernetes . api . v 1 . re conc ilers . ph as ers . Coroutine F link Job P haser Without CR D Ċ import Ġcom . micro focus . fl ork . kubernetes . api . v 1 . re conc ilers . ph as ers . Coroutine F link Job Re conciler P haser Ċ import Ġio . fabric 8 . kubernetes . api . model . Deletion Propagation Ċ import Ġio . fabric 8 . kubernetes . client . Kubernetes Client Ċ import Ġio . fabric 8 . kubernetes . client . inform ers . cache . Cache Ċ import Ġio . fabric 8 . kubernetes . client . inform ers . cache . Lister Ċ import Ġkotlinx . coroutines .* Ċ import Ġorg . apache . flink . configuration . High Availability Options Ċ import Ġorg . slf 4 j . Logger Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . Concurrent Map Ċ import Ġjava . util . concurrent . atomic . Atomic Reference Ċ Ċ class ĠCor outine F link Job Re conciler ( ĊĠĠĠ Ġprivate Ġval Ġk 8 s Client : ĠKubernetes Client , ĊĠĠĠ Ġprivate Ġval Ġlist er : ĠAtomic Reference < Lister < F link Job Custom Resource >? >, ĊĠĠĠ Ġprivate Ġval Ġc rd Based : ĠBoolean Ċ ) Ġ: ĠF link Job Re conciler Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠLOG : ĠLogger Ġ= ĠLoggerFactory . getLogger ( Coroutine F link Job Re conciler :: class . java ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠCOR OUT INE _ SCOPE Ġ= ĠAtomic Reference ( create Coroutine Scope ()) ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠRE CON C IL ER _ STAT ES : ĠConcurrent Map < String , ĠCor outine F link Job Re conciler P haser > Ġ= ĠConcurrent HashMap () ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġcreate Coroutine Scope () Ġ= ĠCor outine Scope ( Dispatch ers . IO Ġ+ ĠSuper visor Job () Ġ+ ĠCor outine ExceptionHandler Ġ{ Ġ_, Ġthrowable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . error (" Error Ġduring ĠF link Job Ġre conc iliation :", Ġthrowable ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġreset Scope () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRE CON C IL ER _ STAT ES . forEach Ġ{ Ġit . value . cancel () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCOR OUT INE _ SCOPE . get AndSet ( create Coroutine Scope ()). cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRE CON C IL ER _ STAT ES . clear () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġmaybe Clean High Availability ( k 8 s Client : ĠKubernetes Client , Ġf link Job : ĠF link Job Custom Resource , Ġkey : ĠString ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( flink Job . spec . flink Conf ?. containsKey ( High Availability Options . HA _ MODE . key ()) Ġ== Ġtrue Ġ&& Ġf link Job . spec . fl ork Conf ?. clean High Availability Ġ== Ġtrue ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . info (" Clean ing ĠHA Ġconfig Ġmaps Ġfor Ġ'{} '. ", Ġkey ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlabels Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" app " Ġto Ġf link Job . metadata . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" config map - type " Ġto Ġ" high - availability ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" type " Ġto Ġ" flink - native - kubernetes " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcm Operation Ġ= Ġk 8 s Client . config Maps () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. in Namespace ( flink Job . metadata . namespace ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Labels ( labels ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġlist Ġ= Ġcm Operation . list () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( list . items ?. isNotEmpty () Ġ== Ġtrue ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . debug (" Deleting Ġ{} Ġconfig Ġmap ( s ). ", Ġlist . items ?. size ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcm Operation . delete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( 1 0 0 L ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlist Ġ= Ġcm Operation . list () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġre concile ( flink Job : ĠF link Job Custom Resource ) Ġ{ ĊĠĠĠĠĠĠĠ ĠCOR OUT INE _ SCOPE . get (). launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= ĠCache . meta Namespace Key Func ( flink Job ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġp haser Candidate Ġ= Ġif Ġ( crd Based ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCor outine F link Job Re conciler P haser ( this , Ġk 8 s Client , Ġlist er , Ġkey ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCor outine F link Job P haser Without CR D ( this , Ġk 8 s Client , Ġlist er , Ġkey ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstate Ġ= ĠRE CON C IL ER _ STAT ES . merge ( key , Ġp haser Candidate ) Ġ{ Ġold , Ġnew Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġold . isActive () Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnew . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġold ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġold . was Generation Ob served ( flink Job ) Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnew . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġold ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnew ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( state ?. start ( this , Ġkey ) Ġ== Ġtrue Ġ|| Ġstate ?. isActive () Ġ== Ġtrue ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate . channel . send ( flink Job ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġdelete ( flink Job : ĠF link Job Custom Resource ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġkey Ġ= ĠCache . meta Namespace Key Func ( flink Job ) ĊĠĠĠĠĠĠĠ ĠRE CON C IL ER _ STAT ES . remove ( key )?. cancel () ĊĊĠĠĠĠĠĠĠ Ġk 8 s Client . apps (). deployments () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. in Namespace ( flink Job . metadata . namespace ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Name ( flink Job . metadata . name ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Propagation Policy ( Deletion Propagation . FO REG ROUND ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. delete () ĊĊĠĠĠĠĠĠĠ Ġmaybe Clean High Availability ( k 8 s Client , Ġf link Job , Ġkey ) ĊĠĠĠ Ġ} Ċ } Ċ