@ Override public void onNext ( final JobSubmissionEvent jobSubmissionEvent ) { LOG . log ( Level . FINEST , "Submitting job: {0}" , jobSubmissionEvent ) ; try { this . applicationId = createApplicationId ( jobSubmissionEvent ) ; final String folderName = this . azureBatchFileNames . getStorageJobFolder ( this . applicationId ) ; LOG . log ( Level . FINE , "Creating a job folder on Azure at: {0}." , folderName ) ; final URI jobFolderURL = this . azureStorageClient . getJobSubmissionFolderUri ( folderName ) ; LOG . log ( Level . FINE , "Getting a shared access signature for {0}." , folderName ) ; final String storageContainerSAS = this . azureStorageClient . createContainerSharedAccessSignature ( ) ; LOG . log ( Level . FINE , "Assembling Configuration for the Driver." ) ; final Configuration driverConfiguration = makeDriverConfiguration ( jobSubmissionEvent , this . applicationId , jobFolderURL ) ; LOG . log ( Level . FINE , "Making Job JAR." ) ; final File jobSubmissionJarFile = this . jobJarMaker . createJobSubmissionJAR ( jobSubmissionEvent , driverConfiguration ) ; LOG . log ( Level . FINE , "Uploading Job JAR to Azure." ) ; final URI jobJarSasUri = this . azureStorageClient . uploadFile ( folderName , jobSubmissionJarFile ) ; LOG . log ( Level . FINE , "Assembling application submission." ) ; final String command = this . launchCommandBuilder . buildDriverCommand ( jobSubmissionEvent ) ; this . azureBatchHelper . submitJob ( getApplicationId ( ) , storageContainerSAS , jobJarSasUri , command ) ; } catch ( final IOException e ) { LOG . log ( Level . SEVERE , "Error submitting Azure Batch request: {0}" , e ) ; throw new RuntimeException ( e ) ; } }
Invoked when JobSubmissionEvent is triggered .