public void adjustJobMetricsOnRetry ( int branches ) { TaskMetrics metrics = TaskMetrics . get ( this ) ; for ( int i = 0 ; i < branches ; i ++ ) { String forkBranchId = ForkOperatorUtils . getForkId ( this . taskId , i ) ; long recordsWritten = metrics . getCounter ( MetricGroup . TASK . name ( ) , forkBranchId , RECORDS ) . getCount ( ) ; long bytesWritten = metrics . getCounter ( MetricGroup . TASK . name ( ) , forkBranchId , BYTES ) . getCount ( ) ; metrics . getCounter ( MetricGroup . JOB . name ( ) , this . jobId , RECORDS ) . dec ( recordsWritten ) ; metrics . getCounter ( MetricGroup . JOB . name ( ) , this . jobId , BYTES ) . dec ( bytesWritten ) ; } }
Adjust job - level metrics when the task gets retried .