public static void run ( JobConfig config , int attempts , AlluxioConfiguration alluxioConf ) throws InterruptedException { CountingRetry retryPolicy = new CountingRetry ( attempts ) ; while ( retryPolicy . attempt ( ) ) { long jobId ; try ( JobMasterClient client = JobMasterClient . Factory . create ( JobMasterClientContext . newBuilder ( ClientContext . create ( alluxioConf ) ) . build ( ) ) ) { jobId = client . run ( config ) ; } catch ( Exception e ) { // job could not be started, retry LOG . warn ( "Exception encountered when starting a job." , e ) ; continue ; } JobInfo jobInfo = waitFor ( jobId , alluxioConf ) ; if ( jobInfo == null ) { // job status could not be fetched, give up break ; } if ( jobInfo . getStatus ( ) == Status . COMPLETED || jobInfo . getStatus ( ) == Status . CANCELED ) { return ; } LOG . warn ( "Job {} failed to complete: {}" , jobId , jobInfo . getErrorMessage ( ) ) ; } throw new RuntimeException ( "Failed to successfully complete the job." ) ; }
Runs the specified job and waits for it to finish . If the job fails it is retried the given number of times . If the job does not complete in the given number of attempts an exception is thrown .