private boolean startRemoteJT ( JobConf jobConf , ResourceGrant grant ) throws InterruptedException { org . apache . hadoop . corona . InetAddress ttAddr = Utilities . appInfoToAddress ( grant . appInfo ) ; CoronaTaskTrackerProtocol coronaTT = null ; try { coronaTT = jt . getTaskTrackerClient ( ttAddr . getHost ( ) , ttAddr . getPort ( ) ) ; } catch ( IOException e ) { LOG . error ( "Error while trying to connect to TT at " + ttAddr . getHost ( ) + ":" + ttAddr . getPort ( ) , e ) ; return false ; } LOG . warn ( "Starting remote JT for " + attemptJobId + " on " + ttAddr . getHost ( ) ) ; // Get a special map id for the JT task. Path systemDir = new Path ( jt . getSystemDir ( ) ) ; LOG . info ( "startRemoteJT:systemDir " + systemDir . toString ( ) ) ; String jobFile = CoronaJobInProgress . getJobFile ( systemDir , attemptJobId ) . toString ( ) ; LOG . info ( "startRemoteJT:jobFile " + jobFile ) ; String splitClass = JobClient . RawSplit . class . getName ( ) ; BytesWritable split = new BytesWritable ( ) ; Task jobTask = new MapTask ( jobFile , currentAttemptId , currentAttemptId . getTaskID ( ) . getId ( ) , splitClass , split , 1 , jobConf . getUser ( ) ) ; CoronaSessionInfo info = new CoronaSessionInfo ( jt . getSessionId ( ) , jt . getJobTrackerAddress ( ) , jt . getJobTrackerAddress ( ) ) ; synchronized ( this ) { try { coronaTT . startCoronaJobTracker ( jobTask , info ) ; } catch ( IOException e ) { // Increment the attempt so that the older attempt will get an error // in reportRemoteCoronaJobTracker(). incrementAttemptUnprotected ( ) ; LOG . error ( "Error while performing RPC to TT at " + ttAddr . getHost ( ) + ":" + ttAddr . getPort ( ) , e ) ; return false ; } } // Now wait for the remote CJT to report its address. final long waitStart = System . currentTimeMillis ( ) ; final long timeout = RemoteJTProxy . getRemotJTTimeout ( jobConf ) ; synchronized ( this ) { while ( client == null ) { LOG . warn ( "Waiting for remote JT to start on " + ttAddr . getHost ( ) ) ; this . wait ( 1000 ) ; if ( client == null && System . currentTimeMillis ( ) - waitStart > timeout ) { // Increment the attempt so that the older attempt will get an error // in reportRemoteCoronaJobTracker(). incrementAttemptUnprotected ( ) ; LOG . warn ( "Could not start remote JT on " + ttAddr . getHost ( ) ) ; return false ; } } } return true ; }
Start corona job tracker on the machine provided by using the corona task tracker API .