public ParallelTask execute ( ParallecResponseHandler handler ) { ParallelTask task = new ParallelTask ( ) ; try { targetHostMeta = new TargetHostMeta ( targetHosts ) ; final ParallelTask taskReal = new ParallelTask ( requestProtocol , concurrency , httpMeta , targetHostMeta , sshMeta , tcpMeta , udpMeta , pingMeta , handler , responseContext , replacementVarMapNodeSpecific , replacementVarMap , requestReplacementType , config ) ; task = taskReal ; logger . info ( "***********START_PARALLEL_HTTP_TASK_" + task . getTaskId ( ) + "***********" ) ; // throws ParallelTaskInvalidException task . validateWithFillDefault ( ) ; task . setSubmitTime ( System . currentTimeMillis ( ) ) ; if ( task . getConfig ( ) . isEnableCapacityAwareTaskScheduler ( ) ) { //late initialize the task scheduler ParallelTaskManager . getInstance ( ) . initTaskSchedulerIfNot ( ) ; // add task to the wait queue ParallelTaskManager . getInstance ( ) . getWaitQ ( ) . add ( task ) ; logger . info ( "Enabled CapacityAwareTaskScheduler. Submitted task to waitQ in builder.. " + task . getTaskId ( ) ) ; } else { logger . info ( "Disabled CapacityAwareTaskScheduler. Immediately execute task {} " , task . getTaskId ( ) ) ; Runnable director = new Runnable ( ) { public void run ( ) { ParallelTaskManager . getInstance ( ) . generateUpdateExecuteTask ( taskReal ) ; } } ; new Thread ( director ) . start ( ) ; } if ( this . getMode ( ) == TaskRunMode . SYNC ) { logger . info ( "Executing task {} in SYNC mode... " , task . getTaskId ( ) ) ; while ( task != null && ! task . isCompleted ( ) ) { try { Thread . sleep ( 500L ) ; } catch ( InterruptedException e ) { logger . error ( "InterruptedException " + e ) ; } } } } catch ( ParallelTaskInvalidException ex ) { logger . info ( "Request is invalid with missing parts. Details: " + ex . getMessage ( ) + " Cannot execute at this time. " + " Please review your request and try again.\nCommand:" + httpMeta . toString ( ) ) ; task . setState ( ParallelTaskState . COMPLETED_WITH_ERROR ) ; task . getTaskErrorMetas ( ) . add ( new TaskErrorMeta ( TaskErrorType . VALIDATION_ERROR , "validation eror" ) ) ; } catch ( Exception t ) { logger . error ( "fail task builder. Unknown error: " + t , t ) ; task . setState ( ParallelTaskState . COMPLETED_WITH_ERROR ) ; task . getTaskErrorMetas ( ) . add ( new TaskErrorMeta ( TaskErrorType . UNKNOWN , "unknown eror" , t ) ) ; } logger . info ( "***********FINISH_PARALLEL_HTTP_TASK_" + task . getTaskId ( ) + "***********" ) ; return task ; }
key function . first validate if the ACM has adequate data ; then execute it after the validation . the new ParallelTask task guareetee to have the targethost meta and command meta not null