private boolean checkDataQuality ( Optional < Object > schema ) throws Exception { if ( this . branches > 1 ) { this . forkTaskState . setProp ( ConfigurationKeys . EXTRACTOR_ROWS_EXPECTED , this . taskState . getProp ( ConfigurationKeys . EXTRACTOR_ROWS_EXPECTED ) ) ; this . forkTaskState . setProp ( ConfigurationKeys . EXTRACTOR_ROWS_EXTRACTED , this . taskState . getProp ( ConfigurationKeys . EXTRACTOR_ROWS_EXTRACTED ) ) ; } String writerRecordsWrittenKey = ForkOperatorUtils . getPropertyNameForBranch ( ConfigurationKeys . WRITER_RECORDS_WRITTEN , this . branches , this . index ) ; if ( this . writer . isPresent ( ) ) { this . forkTaskState . setProp ( ConfigurationKeys . WRITER_ROWS_WRITTEN , this . writer . get ( ) . recordsWritten ( ) ) ; this . taskState . setProp ( writerRecordsWrittenKey , this . writer . get ( ) . recordsWritten ( ) ) ; } else { this . forkTaskState . setProp ( ConfigurationKeys . WRITER_ROWS_WRITTEN , 0L ) ; this . taskState . setProp ( writerRecordsWrittenKey , 0L ) ; } if ( schema . isPresent ( ) ) { this . forkTaskState . setProp ( ConfigurationKeys . EXTRACT_SCHEMA , schema . get ( ) . toString ( ) ) ; } try { // Do task-level quality checking TaskLevelPolicyCheckResults taskResults = this . taskContext . getTaskLevelPolicyChecker ( this . forkTaskState , this . branches > 1 ? this . index : - 1 ) . executePolicies ( ) ; TaskPublisher publisher = this . taskContext . getTaskPublisher ( this . forkTaskState , taskResults ) ; switch ( publisher . canPublish ( ) ) { case SUCCESS : return true ; case CLEANUP_FAIL : this . logger . error ( "Cleanup failed for task " + this . taskId ) ; break ; case POLICY_TESTS_FAIL : this . logger . error ( "Not all quality checking passed for task " + this . taskId ) ; break ; case COMPONENTS_NOT_FINISHED : this . logger . error ( "Not all components completed for task " + this . taskId ) ; break ; default : break ; } return false ; } catch ( Throwable t ) { this . logger . error ( "Failed to check task-level data quality" , t ) ; return false ; } }
Check data quality .