/* Ċ Ġ* ĠLicensed Ġto Ġthe ĠApache ĠSoftware ĠFoundation Ġ( ASF ) Ġunder Ġone Ġor Ġmore Ċ Ġ* Ġcontributor Ġlicense Ġagreements . Ġ ĠSee Ġthe ĠNOTICE Ġfile Ġdistributed Ġwith Ċ Ġ* Ġthis Ġwork Ġfor Ġadditional Ġinformation Ġregarding Ġcopyright Ġownership . Ċ Ġ* ĠThe ĠASF Ġlicenses Ġthis Ġfile Ġto ĠYou Ġ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 . apache . am ater as u . leader . yarn Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġcom . fasterxml . jackson . data format . yaml . YAML Factory Ċ import Ġcom . fasterxml . jackson . module . kotlin . Kotlin Module Ċ Ċ import Ġkotlinx . coroutines . async Ċ import Ġkotlinx . coroutines . run Blocking Ċ Ċ import Ġorg . apache . activ emq . broker . Broker Service Ċ Ċ import Ġorg . apache . am ater as u . common . configuration . Cluster Config Ċ import Ġorg . apache . am ater as u . common . data objects . Action Data Ċ import Ġorg . apache . am ater as u . common . logging . K Logging Ċ import Ġorg . apache . am ater as u . common . utils . Active Notifier Ċ import Ġorg . apache . am ater as u . leader . common . configuration . Config Manager Ċ import Ġorg . apache . am ater as u . leader . common . execution . Job Loader Ċ import Ġorg . apache . am ater as u . leader . common . execution . Job Manager Ċ import Ġorg . apache . am ater as u . leader . common . execution . frameworks . Framework Providers Factory Ċ import Ġorg . apache . am ater as u . leader . common . launcher . A ma Opts Ċ import Ġorg . apache . am ater as u . leader . common . utilities . DataLoader Ċ import Ġorg . apache . am ater as u . leader . common . utilities . Messaging Client Util Ċ import Ġorg . apache . am ater as u . sdk . frameworks . Framework Setup Provider Ċ import Ġorg . apache . am ater as u . sdk . frameworks . Runner Setup Provider Ċ import Ġorg . apache . commons . lang . exception . Exception Utils Ċ Ċ import Ġorg . apache . cur ator . framework . Cur ator Framework Ċ import Ġorg . apache . cur ator . framework . Cur ator Framework Factory Ċ import Ġorg . apache . cur ator . framework . recipes . bar riers . Distributed Barrier Ċ import Ġorg . apache . cur ator . retry . Exponent ial Backoff Retry Ċ Ċ import Ġorg . apache . hadoop . fs . FileSystem Ċ import Ġorg . apache . hadoop . fs . Path Ċ import Ġorg . apache . hadoop . io . Data Output Buffer Ċ import Ġorg . apache . hadoop . security . User Group Information Ċ import Ġorg . apache . hadoop . yarn . api . records .* Ċ import Ġorg . apache . hadoop . yarn . client . api . AM RM Client Ċ import Ġorg . apache . hadoop . yarn . client . api . async . AM RM Client Async Ċ import Ġorg . apache . hadoop . yarn . client . api . async . NM Client Async Ċ import Ġorg . apache . hadoop . yarn . client . api . async . impl . NM Client Async Impl Ċ import Ġorg . apache . hadoop . yarn . conf . Y arn Configuration Ċ import Ġorg . apache . hadoop . yarn . exceptions . Y arn Exception Ċ import Ġorg . apache . hadoop . yarn . security . AM RM Token Identifier Ċ import Ġorg . apache . hadoop . yarn . util . Records Ċ Ċ import Ġorg . apache . zookeeper . Create Mode Ċ import Ġjava . io .* Ċ Ċ import Ġjava . nio . ByteBuffer Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . Concurrent Linked Queue Ċ import Ġjava . util . concurrent . Linked BlockingQueue Ċ Ċ import Ġjavax . jms . Message Consumer Ċ Ċ class ĠApplication Master Ġ: ĠK Logging (), ĠAM RM Client Async . Callback Handler Ġ{ ĊĊĊĠĠĠ Ġlateinit Ġvar Ġaddress : ĠString ĊĊĠĠĠ Ġval Ġbroker : ĠBroker Service Ġ= ĠBroker Service () ĊĠĠĠ Ġprivate Ġval Ġconf Ġ= ĠY arn Configuration () ĊĊĊĠĠĠ Ġprivate Ġval Ġactions Buffer Ġ= ĠConcurrent Linked Queue < Action Data >() ĊĠĠĠ Ġprivate Ġval Ġcompleted Containers And Task Ids Ġ= ĠConcurrent HashMap < Long , ĠString >() ĊĠĠĠ Ġprivate Ġval Ġcontainers Ids To Task Ġ= ĠConcurrent HashMap < Long , ĠAction Data >() ĊĠĠĠ Ġprivate Ġval Ġyaml Mapper Ġ= ĠObjectMapper ( YAML Factory ()) ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġprop Path : ĠString ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġprops : ĠFileInputStream ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġjob Manager : ĠJob Manager ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġzk Client : ĠCur ator Framework ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġenv : ĠString ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġrm Client : ĠAM RM Client Async < AM RM Client . Container Request > ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġframework Factory : ĠFramework Providers Factory ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġconfig : ĠCluster Config ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġfs : ĠFileSystem ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġconsumer : ĠMessage Consumer ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġconfig Manager : ĠConfig Manager ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġnot ifier : ĠActive Notifier ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġnm Client : ĠN M Client Async ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġyaml Mapper . register Module ( Kotlin Module ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġexecute ( opts : ĠA ma Opts ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġprop Path Ġ= ĠSystem . getenv (" PWD ") Ġ+ Ġ"/ am ater as u . properties " ĊĠĠĠĠĠĠĠ Ġprops Ġ= ĠFileInputStream ( File ( prop Path )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġno Ġneed Ġfor ĠH DFS Ġdouble Ġcheck Ġ( nod Ġto ĠA aron ĠRod gers ) ĊĠĠĠĠĠĠĠ Ġ// Ġj ars Ġon ĠH DFS Ġshould Ġhave Ġbeen Ġverified Ġby Ġthe ĠY ARN Ġzk Client ĊĠĠĠĠĠĠĠ Ġconfig Ġ= ĠCluster Config . apply ( props ) ĊĠĠĠĠĠĠĠ Ġfs Ġ= ĠFileSystem . get ( conf ) ĊĊĠĠĠĠĠĠĠ Ġinit Job ( opts ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġnow Ġthat Ġthe Ġjob Ġwas Ġinitiated , Ġthe Ġcur ator Ġzk Client Ġis ĠStarted Ġand Ġwe Ġcan ĊĠĠĠĠĠĠĠ Ġ// Ġregister Ġthe Ġbroker 's Ġaddress ĊĠĠĠĠĠĠĠ Ġzk Client . create (). with Mode ( Create Mode . PER SIST ENT ). for Path ("/ ${ job Manager . jobId }/ broker ") ĊĠĠĠĠĠĠĠ Ġzk Client . setData (). for Path ("/ ${ job Manager . jobId }/ broker ", Ġaddress . toByteArray ()) ĊĊĠĠĠĠĠĠĠ Ġ// Ġonce Ġthe Ġbroker Ġis Ġregistered , Ġwe Ġcan Ġremove Ġthe Ġbarrier Ġso Ġclients Ġcan Ġconnect ĊĠĠĠĠĠĠĠ Ġlog . info ("/ ${ job Manager . jobId }- report - barrier ") ĊĠĠĠĠĠĠĠ Ġval Ġbarrier Ġ= ĠDistributed Barrier ( zk Client , Ġ"/ ${ job Manager . jobId }- report - barrier ") ĊĠĠĠĠĠĠĠ Ġbarrier . remove Barrier () ĊĊĠĠĠĠĠĠĠ Ġconsumer Ġ= ĠMessaging Client Util . setup Messaging ( address ) ĊĠĠĠĠĠĠĠ Ġnot ifier Ġ= ĠActive Notifier ( address ) ĊĊĠĠĠĠĠĠĠ Ġlog . info (" number Ġof Ġmessages Ġ${ broker . admin View . total Message Count }") ĊĊĠĠĠĠĠĠĠ Ġ// ĠInitialize Ġclients Ġto ĠResourceManager Ġand ĠNode Managers ĊĠĠĠĠĠĠĠ Ġnm Client Ġ= ĠN M Client Async Impl ( Y arn NM Callback Handler ( notifier )) ĊĠĠĠĠĠĠĠ Ġnm Client . init ( conf ) ĊĠĠĠĠĠĠĠ Ġnm Client . start () ĊĊĠĠĠĠĠĠĠ Ġrm Client Ġ= Ġstart RM Client () ĊĊĠĠĠĠĠĠĠ Ġval Ġitems Ġ= Ġmutable ListOf < Framework Setup Provider >() ĊĊĠĠĠĠĠĠĠ Ġfor Ġ( p Ġin Ġframework Factory . providers (). values ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġitems . add ( p ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġconfig Items Ġ= Ġitems . flatMap Ġ{ Ġit . configuration Items . as Iterable () Ġ} ĊĠĠĠĠĠĠĠ Ġconfig Manager Ġ= ĠConfig Manager ( env , Ġ" repo ", Ġconfig Items ) ĊĊĊĠĠĠĠĠĠĠ Ġval Ġregistration Response Ġ= Ġrm Client . register Application Master ("", Ġ 0 , Ġ"") ĊĠĠĠĠĠĠĠ Ġval Ġmax Mem Ġ= Ġregistration Response . maximum Resource Capability . memory Size ĊĠĠĠĠĠĠĠ Ġval Ġmax V Co res Ġ= Ġregistration Response . maximum Resource Capability . virtual Co res ĊĊĠĠĠĠĠĠĠ Ġwhile Ġ(! job Manager . out Of Actions ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcapability Ġ= ĠRecords . new Record ( Resource :: class . java ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġaction Data Ġ= Ġjob Manager . next Action Data ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( action Data Ġ!= Ġnull ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnot ifier . info (" request ing Ġcontainer Ġfo Ġ${ action Data . name }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġframework Provider Ġ= Ġframework Factory . get Framework ( action Data . groupId ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdriver Configuration Ġ= Ġframework Provider . driver Configuration ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġmem : ĠLong Ġ= Ġdriver Configuration . memory . to Long () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmem Ġ= ĠMath . min ( mem , Ġmax Mem ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcapability . memory Size Ġ= Ġmem ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġcpu Ġ= Ġdriver Configuration . cpus ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcpu Ġ= ĠMath . min ( cpu , Ġmax V Co res ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcapability . virtual Co res Ġ= Ġcpu ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Task Configuration ( action Data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Container ( action Data , Ġcapability ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġlog . info (" Finished Ġrequesting Ġcontainers ") ĊĠĠĠĠĠĠĠ Ġread Line () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġinit Job ( opts : ĠA ma Opts ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġthis . env Ġ= Ġopts . env ĊĠĠĠĠĠĠĠ Ġframework Factory Ġ= ĠFramework Providers Factory . apply ( env , Ġconfig ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġretry Policy Ġ= ĠExp onent ial Backoff Retry ( 1 0 0 0 , Ġ 3 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġzk Client Ġ= ĠCur ator Framework Factory . new Client ( config . zk (), Ġretry Policy ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġzk Client . start () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Error Ġconnecting Ġto Ġz ookeeper ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġzk Path Ġ= Ġzk Client . check Exists (). for Path ("/ ${ opts . new JobId }") ĊĊĠĠĠĠĠĠĠ Ġlog . info (" zk Path Ġis Ġ$ zk Path ") ĊĠĠĠĠĠĠĠ Ġif Ġ( zk Path Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" res uming Ġjob " Ġ+ Ġopts . new JobId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjob Manager Ġ= ĠJob Loader . reload Job ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . new JobId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . userName , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . password , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġzk Client , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconfig . Jobs (). tasks (). attempts (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLinked BlockingQueue < Action Data >()) ĊĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" new Ġjob Ġis Ġbeing Ġcreated ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob Manager Ġ= ĠJob Loader . load Job ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . repo , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . branch , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . new JobId , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . userName , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġopts . password , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġzk Client , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconfig . Jobs (). tasks (). attempts (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLinked BlockingQueue < Action Data >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Error Ġcreating ĠJob Manager .", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġjob Manager . start () ĊĠĠĠĠĠĠĠ Ġlog . info (" Started Ġjob Manager ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Containers Allocated ( containers : ĠMutable List < Container > ?) Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġcontainers ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( container Ġin Ġit ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" container Ġ${ container . id } Ġallocated ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( actions Buffer . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġaction Data Ġ= Ġactions Buffer . poll () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// val Ġcd Ġ= Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" container Ġ${ container . id } Ġallocated ") ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġframework Ġ= Ġframework Factory . get Framework ( action Data . groupId ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrunner Provider Ġ= Ġframework . get Runner Provider ( action Data . type Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġctx Ġ= ĠRecords . new Record ( Container Launch Context :: class . java ) ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġenv Conf Ġ= Ġconfig Manager . getAction Configuration ( action Data . name , Ġaction Data . config ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommands : ĠList < String > Ġ= ĠlistOf ( runner Provider . get Command ( job Manager . jobId , Ġaction Data , Ġenv Conf , Ġ"${ action Data . id }-${ container . id . container Id }", Ġaddress )) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnot ifier . info (" container Ġcommand Ġ${ commands . join ToString ( prefix Ġ= Ġ" Ġ", Ġpostfix Ġ= Ġ" Ġ" )} ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġctx . commands Ġ= Ġcommands ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġctx . tokens Ġ= Ġall Tokens () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġctx . local Resources Ġ= Ġsetup Container Resources ( framework , Ġrunner Provider , Ġaction Data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġctx . environment Ġ= Ġframework . environment Variables ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnm Client . start Container Async ( container , Ġctx ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob Manager . action Started ( action Data . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontainers Ids To Task [ container . id . container Id ] Ġ= Ġaction Data ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnot ifier . info (" created Ġcontainer Ġfor Ġ${ action Data . name } Ġcreated ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ctx . local Resources . forEach Ġ{ Ġt : ĠString , Ġu : ĠLocal Resource Ġ-> Ġnot ifier . info (" resource : Ġ$ t Ġ= Ġ${ u . resource }") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" launch ing Ġcontainer Ġsucceeded : Ġ${ container . id . container Id }; Ġtask : Ġ${ action Data . id }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnot ifier . error ("", Ġ" error Ġlaunch ing Ġcontainer Ġwith Ġ${ e . message } Ġin Ġ${ Exception Utils . get StackTrace ( e )} ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Container ( action Data , Ġcontainer . resource ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} !! ĊĊĠĠĠ Ġprivate Ġfun Ġall Tokens (): ĠByteBuffer Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġcreating Ġthe Ġcredentials Ġfor Ġcontainer Ġexecution ĊĠĠĠĠĠĠĠ Ġval Ġcredentials Ġ= ĠUser Group Information . getCurrent User (). credentials ĊĠĠĠĠĠĠĠ Ġval Ġdo b Ġ= ĠData Output Buffer () ĊĠĠĠĠĠĠĠ Ġcredentials . write Token Storage To Stream ( dob ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġremoving Ġthe ĠAM -> RM Ġtoken Ġso Ġthat Ġcontainers Ġcannot Ġaccess Ġit . ĊĠĠĠĠĠĠĠ Ġval Ġiter Ġ= Ġcredentials . all Tokens . iterator () ĊĠĠĠĠĠĠĠ Ġlog . info (" Executing Ġwith Ġtokens :") ĊĠĠĠĠĠĠĠ Ġfor Ġ( token Ġin Ġiter ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info ( token . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( token . kind Ġ== ĠAM RM Token Identifier . KIND _ NAME ) Ġiter . remove () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠByteBuffer . wrap ( dob . data , Ġ 0 , Ġdo b . length ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCreates Ġthe Ġmap Ġof Ġresources Ġto Ġbe Ġcopied Ġinto Ġthe Ġcontainer ĊĠĠĠĠ Ġ* Ġ@ framework ĠThe Ġframework Setup Provider Ġfor Ġthe Ġaction ĊĠĠĠĠ Ġ* Ġ@ runner Provider Ġthe Ġactions Ġrunner Ġprovider ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġsetup Container Resources ( framework : ĠFramework Setup Provider , Ġrunner Provider : ĠRunner Setup Provider , Ġaction Data : ĠAction Data ): ĠMap < String , ĠLocal Resource > Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġyarn Jar Path Ġ= ĠPath ( config . yarn (). hdfs J ars Path ()) ĊĊĠĠĠĠĠĠĠ Ġ// ĠGetting Ġframework Ġ( group ) Ġresources ĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġframework . group Resources . map Ġ{ Ġit . path Ġto Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , Ġcreate Dist Path ( it . path ))) Ġ}. toMap (). to Mutable Map () ĊĊĠĠĠĠĠĠĠ Ġ// ĠGetting Ġrunner Ġresources ĊĠĠĠĠĠĠĠ Ġresult . put All ( runner Provider . runner Resources . map Ġ{ Ġit Ġto Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , Ġcreate Dist Path ( it ))) Ġ}. toMap ()) ĊĊĠĠĠĠĠĠĠ Ġ// Ġgetting Ġthe Ġaction Ġspecific Ġresources ĊĠĠĠĠĠĠĠ Ġresult . put All ( runner Provider . getAction Resources ( job Manager . jobId , Ġaction Data ). map Ġ{ Ġit . remove Prefix ("${ job Manager . jobId }/${ action Data . name }/ ") Ġto Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , Ġcreate Dist Path ( it ))) Ġ}) ĊĊĠĠĠĠĠĠĠ Ġ// Ġgetting Ġthe Ġaction Ġspecific Ġdependencies ĊĠĠĠĠĠĠĠ Ġrunner Provider . getAction Dependencies ( job Manager . jobId , Ġaction Data ). forEach Ġ{ Ġdistribute File ( it , Ġ"${ job Manager . jobId }/${ action Data . name }/ ") Ġ} ĊĊĠĠĠĠĠĠĠ Ġresult . put All ( runner Provider . getAction Dependencies ( job Manager . jobId , Ġaction Data ). map Ġ{ ĠFile ( it ). name Ġto Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , Ġcreate Dist Path ("${ job Manager . jobId }/${ action Data . name }/ $ it "))) Ġ}) ĊĊĠĠĠĠĠĠĠ Ġ// ĠAdding Ġthe ĠA mat eras u Ġconfiguration Ġfiles ĊĠĠĠĠĠĠĠ Ġresult [" am ater as u . properties "] Ġ= Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , ĠPath ("/ am ater as u . properties "))) ĊĠĠĠĠĠĠĠ Ġresult [" log 4 j . properties "] Ġ= Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , ĠPath ("/ log 4 j . properties "))) ĊĊĠĠĠĠĠĠĠ Ġ// Ġgetting Ġthe Ġaction Ġexecutable ĊĠĠĠĠĠĠĠ Ġval Ġexecutable Ġ= Ġrunner Provider . getAction Executable ( job Manager . jobId , Ġaction Data ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġsetting Ġthe Ġaction Ġexecutable ĊĠĠĠĠĠĠĠ Ġdistribute File ( executable , Ġ"${ job Manager . jobId }/${ action Data . name }/ ") ĊĠĠĠĠĠĠĠ Ġresult [ File ( executable ). name ] Ġ= Ġcreate Local Resource From Path ( Path . merge Paths ( yarn Jar Path , Ġcreate Dist Path ("${ job Manager . jobId }/${ action Data . name }/ $ executable "))) ĊĊĠĠĠĠĠĠĠ Ġresult . forEach Ġ{ Ġlog . debug (" entry Ġ${ it . key } Ġwith Ġvalue Ġ${ it . value }") Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġresult . map Ġ{ Ġx Ġ-> Ġx . key . remove Prefix ("/") Ġto Ġx . value Ġ}. toMap () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Task Configuration ( action Data : ĠAction Data ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ// Ġsetting Ġup Ġthe Ġconfiguration Ġfiles Ġfor Ġthe Ġcontainer ĊĠĠĠĠĠĠĠ Ġval Ġenv Yaml Ġ= Ġconfig Manager . getAction Config Content ( action Data . name , Ġaction Data . config ) ĊĠĠĠĠĠĠĠ Ġwrite ConfigFile ( env Yaml , Ġjob Manager . jobId , Ġaction Data . name , Ġ" env . yaml ") ĊĊĠĠĠĠĠĠĠ Ġval Ġdata Stores Ġ= ĠDataLoader . get Task Data ( action Data , Ġenv ). exports ĊĠĠĠĠĠĠĠ Ġval Ġdata Stores Yaml Ġ= Ġyaml Mapper . writeValue AsString ( data Stores ) ĊĠĠĠĠĠĠĠ Ġwrite ConfigFile ( data Stores Yaml , Ġjob Manager . jobId , Ġaction Data . name , Ġ" data stores . yaml ") ĊĊĠĠĠĠĠĠĠ Ġval Ġdatasets Ġ= ĠDataLoader . getData sets ( env ) ĊĠĠĠĠĠĠĠ Ġwrite ConfigFile ( datasets , Ġjob Manager . jobId , Ġaction Data . name , Ġ" datasets . yaml ") ĊĊĠĠĠĠĠĠĠ Ġwrite ConfigFile (" jobId : Ġ${ job Manager . jobId }\ n action Name : Ġ${ action Data . name }", Ġjob Manager . jobId , Ġaction Data . name , Ġ" runtime . yaml ") ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġwrite ConfigFile ( content : ĠString , Ġjob Id : ĠString , Ġaction Name : ĠString , ĠfileName : ĠString ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġaction Dist Path Ġ= Ġcreate Dist Path ("$ jobId /$ action Name /$ fileName ") ĊĠĠĠĠĠĠĠ Ġval Ġyarn Jar Path Ġ= ĠPath ( config . yarn (). hdfs J ars Path ()) ĊĠĠĠĠĠĠĠ Ġval Ġtarget Path Ġ= ĠPath . merge Paths ( yarn Jar Path , Ġaction Dist Path ) ĊĊĠĠĠĠĠĠĠ Ġval ĠoutputStream Ġ= Ġfs . create ( target Path ) ĊĠĠĠĠĠĠĠ ĠoutputStream . write ( content . toByteArray ()) ĊĠĠĠĠĠĠĠ ĠoutputStream . close () ĊĠĠĠĠĠĠĠ Ġlog . info (" written Ġfile Ġ$ target Path ") ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġdistribute File ( file : ĠString , Ġdistribution Path : ĠString ) Ġ{ ĊĊĊĠĠĠĠĠĠĠ Ġval Ġaction Dist Path Ġ= Ġcreate Dist Path ("$ distribution Path /$ file ") ĊĠĠĠĠĠĠĠ Ġval Ġyarn Jar Path Ġ= ĠPath ( config . yarn (). hdfs J ars Path ()) ĊĠĠĠĠĠĠĠ Ġval Ġtarget Path Ġ= ĠPath . merge Paths ( yarn Jar Path , Ġaction Dist Path ) ĊĊĠĠĠĠĠĠĠ Ġnot ifier . info (" copying Ġfile Ġ$ file , Ġfile Ġstatus Ġ${ File ( file ). exists ()} Ġto Ġ$ target Path ") ĊĊĠĠĠĠĠĠĠ Ġlog . info (" target Ġis Ġ$ target Path ") ĊĊĠĠĠĠĠĠĠ Ġfs . copyFrom Local File ( false , Ġtrue , ĠPath ( file ), Ġtarget Path ) ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Dist Path ( path : ĠString ): ĠPath Ġ= ĠPath ("/ dist /$ path ") ĊĊĠĠĠ Ġprivate Ġfun Ġstart RM Client (): ĠAM RM Client Async < AM RM Client . Container Request > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclient Ġ= ĠAM RM Client Async . create AM RM Client Async < AM RM Client . Container Request >( 1 0 0 0 , Ġthis ) ĊĠĠĠĠĠĠĠ Ġclient . init ( conf ) ĊĠĠĠĠĠĠĠ Ġclient . start () ĊĠĠĠĠĠĠĠ Ġreturn Ġclient ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Local Resource From Path ( path : ĠPath ): ĠLocal Resource Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġstat Ġ= Ġfs . getFile Status ( path ) ĊĠĠĠĠĠĠĠ Ġval Ġfile Resource Ġ= ĠRecords . new Record ( Local Resource :: class . java ) ĊĊĠĠĠĠĠĠĠ Ġfile Resource . shouldBe Uploaded To Shared Cache Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġfile Resource . visibility Ġ= ĠLocal Resource Visibility . PUBLIC ĊĠĠĠĠĠĠĠ Ġfile Resource . resource Ġ= ĠURL . from Path ( path ) ĊĠĠĠĠĠĠĠ Ġfile Resource . size Ġ= Ġstat . len ĊĠĠĠĠĠĠĠ Ġfile Resource . timestamp Ġ= Ġstat . modification Time ĊĠĠĠĠĠĠĠ Ġfile Resource . type Ġ= ĠLocal ResourceType . FILE ĊĠĠĠĠĠĠĠ Ġfile Resource . visibility Ġ= ĠLocal Resource Visibility . PUBLIC ĊĠĠĠĠĠĠĠ Ġreturn Ġfile Resource ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġrequest Container ( action Data : ĠAction Data , Ġcapability : ĠResource ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġactions Buffer . add ( action Data ) ĊĠĠĠĠĠĠĠ Ġlog . info (" About Ġto Ġask Ġcontainer Ġfor Ġaction Ġ${ action Data . id } Ġwith Ġmem Ġ${ capability . memory } Ġand Ġcores Ġ${ capability . virtual Co res }. ĠAction Ġbuffer Ġsize Ġis : Ġ${ actions Buffer . size }") ĊĊĠĠĠĠĠĠĠ Ġ// Ġwe Ġhave Ġan Ġaction Ġto Ġschedule , Ġlet 's Ġrequest Ġa Ġcontainer ĊĠĠĠĠĠĠĠ Ġval Ġpriority : ĠPriority Ġ= ĠRecords . new Record ( Priority :: class . java ) ĊĠĠĠĠĠĠĠ Ġpriority . priority Ġ= Ġ 1 ĊĠĠĠĠĠĠĠ Ġval Ġcontainer Req Ġ= ĠAM RM Client . Container Request ( capability , Ġnull , Ġnull , Ġpriority ) ĊĠĠĠĠĠĠĠ Ġrm Client . add Container Request ( container Req ) ĊĠĠĠĠĠĠĠ Ġlog . info (" Ask ed Ġcontainer Ġfor Ġaction Ġ${ action Data . id }") ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Nodes Updated ( updated Nodes : ĠMutable List < Node Report > ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . info (" Nodes Ġchange . ĠNothing Ġto Ġreport .") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Shutdown Request () Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . error (" Shutdown Ġrequested .") ĊĠĠĠĠĠĠĠ Ġstop Application ( Final Application Status . K IL LED , Ġ" Shutdown Ġrequested ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Progress (): ĠFloat Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġjob Manager . registered Actions . size . toFloat () Ġ/ Ġcompleted Containers And Task Ids . size ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonError ( e : ĠThrowable ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġnot ifier . error (" Error Ġrunning Ġa Ġcontainer Ġ${ e !!. message !! }", ĠException Utils . get StackTrace ( e )) ĊĠĠĠĠĠĠĠ Ġstop Application ( Final Application Status . FAILED , Ġ" Error Ġon ĠAM ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Containers Completed ( statuses : ĠMutable List < Container Status > ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġfor Ġ( status Ġin Ġstatuses !! ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( status . state Ġ== ĠContainer State . COMPLETE ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcontainer Id Ġ= Ġstatus . container Id . container Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtask Ġ= Ġcontainers Ids To Task [ container Id ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrm Client . release Assigned Container ( status . container Id ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠtaskId Ġ= Ġtask !!. id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( status . exit Status Ġ== Ġ 0 ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// completed Containers And Task Ids . put ( container Id , Ġtask . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob Manager . action Complete ( taskId ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnot ifier . info (" Container Ġ$ container Id ĠComplete Ġwith Ġtask Ġ$ taskId Ġwith Ġsuccess .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO : ĠCheck Ġthe Ġget Diagnostics Ġvalue Ġand Ġsee Ġif Ġappropriate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob Manager . action Failed ( taskId , Ġstatus . diagnostics ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnot ifier . error (" Container Ġ$ container Id ĠComplete Ġwith Ġtask Ġ$ taskId Ġwith ĠFailed Ġstatus Ġcode Ġ(${ status . exit Status })", Ġstatus . diagnostics ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( job Manager . out Of Actions ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Finished Ġall Ġtasks Ġsuccessfully ! ĠW ow !") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjob Manager . actions Count () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstop Application ( Final Application Status . SUCCEEDED , Ġ" SUCCESS ") ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" job Manager . registered Actions . size : Ġ${ job Manager . registered Actions . size }; Ġcompleted Containers And Task Ids . size : Ġ${ completed Containers And Task Ids . size }") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġstop Application ( final Application Status : ĠFinal Application Status , Ġapp Message : ĠString ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrm Client . unregister Application Master ( final Application Status , Ġapp Message , Ġnull ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠY arn Exception ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġunregister Ġapplication ", Ġex ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġunregister Ġapplication ", Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġrm Client . stop () ĊĠĠĠĠĠĠĠ Ġnm Client . stop () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġmain ( args : ĠArray < String >) Ġ= ĠApp Master Args Parser (). main ( args ) ĊĊĠĠĠ Ġ} Ċ } Ċ