void commit ( final boolean isJobCancelled ) throws IOException { this . datasetStatesByUrns = Optional . of ( computeDatasetStatesByUrns ( ) ) ; final boolean shouldCommitDataInJob = shouldCommitDataInJob ( this . jobState ) ; final DeliverySemantics deliverySemantics = DeliverySemantics . parse ( this . jobState ) ; final int numCommitThreads = numCommitThreads ( ) ; if ( ! shouldCommitDataInJob ) { this . logger . info ( "Job will not commit data since data are committed by tasks." ) ; } try { if ( this . datasetStatesByUrns . isPresent ( ) ) { this . logger . info ( "Persisting dataset urns." ) ; this . datasetStateStore . persistDatasetURNs ( this . jobName , this . datasetStatesByUrns . get ( ) . keySet ( ) ) ; } List < Either < Void , ExecutionException > > result = new IteratorExecutor <> ( Iterables . transform ( this . datasetStatesByUrns . get ( ) . entrySet ( ) , new Function < Map . Entry < String , DatasetState > , Callable < Void > > ( ) { @ Nullable @ Override public Callable < Void > apply ( final Map . Entry < String , DatasetState > entry ) { return createSafeDatasetCommit ( shouldCommitDataInJob , isJobCancelled , deliverySemantics , entry . getKey ( ) , entry . getValue ( ) , numCommitThreads > 1 , JobContext . this ) ; } } ) . iterator ( ) , numCommitThreads , ExecutorsUtils . newThreadFactory ( Optional . of ( this . logger ) , Optional . of ( "Commit-thread-%d" ) ) ) . executeAndGetResults ( ) ; IteratorExecutor . logFailures ( result , LOG , 10 ) ; if ( ! IteratorExecutor . verifyAllSuccessful ( result ) ) { this . jobState . setState ( JobState . RunningState . FAILED ) ; throw new IOException ( "Failed to commit dataset state for some dataset(s) of job " + this . jobId ) ; } } catch ( InterruptedException exc ) { throw new IOException ( exc ) ; } this . jobState . setState ( JobState . RunningState . COMMITTED ) ; }
Commit the job based on whether the job is cancelled .