code: 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);
      }
    }
docs: Adjust job-level metrics when the task gets retried.
  
  @param branches number of forked branches