private ListenableFuture < TaskStatus > attachCallbacks ( final Task task , final ListenableFuture < TaskStatus > statusFuture ) { final ServiceMetricEvent . Builder metricBuilder = new ServiceMetricEvent . Builder ( ) ; IndexTaskUtils . setTaskDimensions ( metricBuilder , task ) ; Futures . addCallback ( statusFuture , new FutureCallback < TaskStatus > ( ) { @ Override public void onSuccess ( final TaskStatus status ) { log . info ( "Received %s status for task: %s" , status . getStatusCode ( ) , status . getId ( ) ) ; handleStatus ( status ) ; } @ Override public void onFailure ( final Throwable t ) { log . makeAlert ( t , "Failed to run task" ) . addData ( "task" , task . getId ( ) ) . addData ( "type" , task . getType ( ) ) . addData ( "dataSource" , task . getDataSource ( ) ) . emit ( ) ; handleStatus ( TaskStatus . failure ( task . getId ( ) ) ) ; } private void handleStatus ( final TaskStatus status ) { try { // If we're not supposed to be running anymore, don't do anything. Somewhat racey if the flag gets set // after we check and before we commit the database transaction, but better than nothing. if ( ! active ) { log . info ( "Abandoning task due to shutdown: %s" , task . getId ( ) ) ; return ; } notifyStatus ( task , status , "notified status change from task" ) ; // Emit event and log, if the task is done if ( status . isComplete ( ) ) { IndexTaskUtils . setTaskStatusDimensions ( metricBuilder , status ) ; emitter . emit ( metricBuilder . build ( "task/run/time" , status . getDuration ( ) ) ) ; log . info ( "Task %s: %s (%d run duration)" , status . getStatusCode ( ) , task , status . getDuration ( ) ) ; if ( status . isSuccess ( ) ) { Counters . incrementAndGetLong ( totalSuccessfulTaskCount , task . getDataSource ( ) ) ; } else { Counters . incrementAndGetLong ( totalFailedTaskCount , task . getDataSource ( ) ) ; } } } catch ( Exception e ) { log . makeAlert ( e , "Failed to handle task status" ) . addData ( "task" , task . getId ( ) ) . addData ( "statusCode" , status . getStatusCode ( ) ) . emit ( ) ; } } } ) ; return statusFuture ; }
Attach success and failure handlers to a task status future such that when it completes we perform the appropriate updates .