@ SuppressWarnings ( "deprecation" ) public ResponseFromManager sendTaskToExecutionManager ( ParallelTask task ) { ResponseFromManager commandResponseFromManager = null ; ActorRef executionManager = null ; try { // Start new job logger . info ( "!!STARTED sendAgentCommandToManager : " + task . getTaskId ( ) + " at " + PcDateUtils . getNowDateTimeStr ( ) ) ; executionManager = ActorConfig . createAndGetActorSystem ( ) . actorOf ( Props . create ( ExecutionManager . class , task ) , "ExecutionManager-" + task . getTaskId ( ) ) ; final FiniteDuration duration = Duration . create ( task . getConfig ( ) . getTimeoutAskManagerSec ( ) , TimeUnit . SECONDS ) ; // Timeout timeout = new // Timeout(FiniteDuration.parse("300 seconds")); Future < Object > future = Patterns . ask ( executionManager , new InitialRequestToManager ( task ) , new Timeout ( duration ) ) ; // set ref task . executionManager = executionManager ; commandResponseFromManager = ( ResponseFromManager ) Await . result ( future , duration ) ; logger . info ( "!!COMPLETED sendTaskToExecutionManager : " + task . getTaskId ( ) + " at " + PcDateUtils . getNowDateTimeStr ( ) + " \t\t GenericResponseMap in future size: " + commandResponseFromManager . getResponseCount ( ) ) ; } catch ( Exception ex ) { logger . error ( "Exception in sendTaskToExecutionManager {} details {}: " , ex , ex ) ; } finally { // stop the manager if ( executionManager != null && ! executionManager . isTerminated ( ) ) { ActorConfig . createAndGetActorSystem ( ) . stop ( executionManager ) ; } if ( task . getConfig ( ) . isAutoSaveLogToLocal ( ) ) { task . saveLogToLocal ( ) ; } } return commandResponseFromManager ; }
Send parallel task to execution manager .